i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 09:20:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/pr/544'

Browse Source
This commit is contained in:
fawney19
2026-05-23 18:41:52 +08:00
parent 6447fda852 781830a202
commit 4b66cadf15
36 changed files with 4329 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apps/aether-gateway/Cargo.toml
View File

@@ -57,6 +57,7 @@ serde.workspace = true
serde_json.workspace = true
sha1 = "0.10"
sha2 = { workspace = true, features = ["oid"] }
tar.workspace = true
sqlx.workspace = true
thiserror.workspace = true
tokio.workspace = true

7
apps/aether-gateway/build.rs
View File

@@ -3,6 +3,7 @@ use std::process::Command;
fn main() {
println!("cargo:rerun-if-env-changed=AETHER_BUILD_VERSION");
println!("cargo:rerun-if-env-changed=AETHER_BUILD_TYPE");
println!("cargo:rerun-if-env-changed=AETHER_VERSION");
println!("cargo:rerun-if-env-changed=GITHUB_REF_NAME");
println!("cargo:rerun-if-changed=../../.git/HEAD");
@@ -27,6 +28,12 @@ fn main() {
.unwrap_or(package_version);
println!("cargo:rustc-env=AETHER_BUILD_VERSION={version}");
let build_type = env::var("AETHER_BUILD_TYPE")
.ok()
.filter(|value| !value.trim().is_empty())
.unwrap_or_else(|| "source".to_string());
println!("cargo:rustc-env=AETHER_BUILD_TYPE={build_type}");
}
fn git_describe_version() -> Option<String> {

59
apps/aether-gateway/src/control/route/admin/system_families.rs
View File

@@ -23,6 +23,65 @@ pub(super) fn classify_admin_system_family_route(
"admin:system",
false,
))
} else if method == http::Method::GET && normalized_path == "/api/admin/system/releases" {
Some(classified(
"admin_proxy",
"system_manage",
"releases",
"admin:system",
false,
))
} else if method == http::Method::GET
&& normalized_path == "/api/admin/system/update-capability"
{
Some(classified(
"admin_proxy",
"system_manage",
"update_capability",
"admin:system",
false,
))
} else if method == http::Method::POST && normalized_path == "/api/admin/system/prepare-update"
{
Some(classified(
"admin_proxy",
"system_manage",
"prepare_update",
"admin:system",
false,
))
} else if method == http::Method::POST && normalized_path == "/api/admin/system/apply-update" {
Some(classified(
"admin_proxy",
"system_manage",
"apply_update",
"admin:system",
false,
))
} else if method == http::Method::POST && normalized_path == "/api/admin/system/rollback" {
Some(classified(
"admin_proxy",
"system_manage",
"rollback",
"admin:system",
false,
))
} else if method == http::Method::GET && normalized_path == "/api/admin/system/update-status" {
Some(classified(
"admin_proxy",
"system_manage",
"update_status",
"admin:system",
false,
))
} else if method == http::Method::GET && normalized_path == "/api/admin/system/update-history" {
Some(classified(
"admin_proxy",
"system_manage",
"update_history",
"admin:system",
false,
))
} else if method == http::Method::GET && normalized_path == "/api/admin/system/aws-regions" {
Some(classified(
"admin_proxy",

49
apps/aether-gateway/src/control/tests/admin_core.rs
View File

@@ -238,6 +238,55 @@ fn classifies_admin_system_check_update_as_admin_proxy_route() {
assert!(!decision.is_execution_runtime_candidate());
}
#[test]
fn classifies_admin_system_update_routes_as_admin_proxy_routes() {
let headers = headers(&[]);
let cases = [
(
http::Method::GET,
"/api/admin/system/update-capability",
"update_capability",
),
(
http::Method::POST,
"/api/admin/system/prepare-update",
"prepare_update",
),
(
http::Method::POST,
"/api/admin/system/apply-update",
"apply_update",
),
(http::Method::POST, "/api/admin/system/rollback", "rollback"),
(http::Method::GET, "/api/admin/system/releases", "releases"),
(
http::Method::GET,
"/api/admin/system/update-history",
"update_history",
),
(
http::Method::GET,
"/api/admin/system/update-status",
"update_status",
),
];
for (method, path, expected_kind) in cases {
let uri: Uri = path.parse().expect("uri should parse");
let decision =
classify_control_route(&method, &uri, &headers).expect("route should classify");
assert_eq!(decision.route_class.as_deref(), Some("admin_proxy"));
assert_eq!(decision.route_family.as_deref(), Some("system_manage"));
assert_eq!(decision.route_kind.as_deref(), Some(expected_kind));
assert_eq!(
decision.auth_endpoint_signature.as_deref(),
Some("admin:system")
);
assert!(!decision.is_execution_runtime_candidate());
}
}
#[test]
fn classifies_admin_system_aws_regions_as_admin_proxy_route() {
let headers = headers(&[]);

139
apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs
View File

@@ -13,10 +13,16 @@ use crate::handlers::admin::system::shared::paths::{
};
use crate::handlers::admin::system::shared::settings::{
apply_admin_system_settings_update, build_admin_api_formats_payload,
build_admin_system_check_update_payload_from_release, build_admin_system_settings_payload,
build_admin_system_stats_payload, current_aether_version, fetch_latest_admin_system_release,
build_admin_system_check_update_payload_from_release, build_admin_system_releases_list_payload,
build_admin_system_settings_payload, build_admin_system_stats_payload, current_aether_version,
fetch_admin_system_releases, fetch_latest_admin_system_release, resolve_update_target,
};
use crate::handlers::admin::system::shared::smtp::build_admin_smtp_test_payload;
use crate::handlers::admin::system::shared::update::{
build_admin_system_update_capability_payload, prepare_admin_system_update_task,
read_update_history, read_update_task_status, start_admin_system_rollback_task,
start_admin_system_update_task,
};
use crate::important_notification::build_important_notification_test_payload;
use crate::maintenance::{ManualUsageCleanupMode, ManualUsageCleanupOptions};
use crate::GatewayError;
@@ -58,7 +64,8 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/check-update"
{
let (latest_release, error) = fetch_latest_admin_system_release().await;
let force = query_flag(request_context.query_string(), "force");
let (latest_release, error) = fetch_latest_admin_system_release(force).await;
return Ok(Some(
Json(build_admin_system_check_update_payload_from_release(
latest_release,
@@ -68,6 +75,123 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
));
}
if decision.route_kind.as_deref() == Some("releases")
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/releases"
{
let force = query_flag(request_context.query_string(), "force");
let (releases, error) = fetch_admin_system_releases(force).await;
return Ok(Some(
Json(build_admin_system_releases_list_payload(releases, error)).into_response(),
));
}
if decision.route_kind.as_deref() == Some("update_capability")
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/update-capability"
{
return Ok(Some(
Json(build_admin_system_update_capability_payload()).into_response(),
));
}
if decision.route_kind.as_deref() == Some("prepare_update")
&& request_method == http::Method::POST
&& request_path == "/api/admin/system/prepare-update"
{
let target_version = request_body
.filter(|b| !b.is_empty())
.and_then(|body| serde_json::from_slice::<serde_json::Value>(body).ok())
.and_then(|v| v.get("version").and_then(|v| v.as_str().map(String::from)));
let (version, tarball_url, sha256sums_url) =
match resolve_update_target(target_version).await {
Ok(result) => result,
Err((status, payload)) => {
return Ok(Some((status, Json(payload)).into_response()));
}
};
return Ok(Some(
match prepare_admin_system_update_task(version, tarball_url, sha256sums_url).await? {
Ok(payload) => attach_admin_audit_response(
Json(payload).into_response(),
"admin_system_update_prepared",
"prepare_system_update",
"system_update",
"global",
),
Err((status, payload)) => (status, Json(payload)).into_response(),
},
));
}
if decision.route_kind.as_deref() == Some("apply_update")
&& request_method == http::Method::POST
&& request_path == "/api/admin/system/apply-update"
{
let version = request_body
.filter(|b| !b.is_empty())
.and_then(|body| serde_json::from_slice::<serde_json::Value>(body).ok())
.and_then(|v| v.get("version").and_then(|v| v.as_str().map(String::from)));
return Ok(Some(
match start_admin_system_update_task(version).await? {
Ok(payload) => attach_admin_audit_response(
Json(payload).into_response(),
"admin_system_update_started",
"apply_system_update",
"system_update",
"global",
),
Err((status, payload)) => (status, Json(payload)).into_response(),
},
));
}
if decision.route_kind.as_deref() == Some("rollback")
&& request_method == http::Method::POST
&& request_path == "/api/admin/system/rollback"
{
return Ok(Some(match start_admin_system_rollback_task().await? {
Ok(payload) => attach_admin_audit_response(
Json(payload).into_response(),
"admin_system_rollback_started",
"rollback_system_update",
"system_rollback",
"global",
),
Err((status, payload)) => (status, Json(payload)).into_response(),
}));
}
if decision.route_kind.as_deref() == Some("update_status")
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/update-status"
{
let status = read_update_task_status();
return Ok(Some(
Json(json!({
"phase": status.phase,
"error": status.error,
"output": status.output,
"progress_label": status.progress_label,
"downloaded_bytes": status.downloaded_bytes,
"total_bytes": status.total_bytes,
"progress_percent": status.progress_percent,
}))
.into_response(),
));
}
if decision.route_kind.as_deref() == Some("update_history")
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/update-history"
{
let entries = read_update_history();
return Ok(Some(Json(json!({ "entries": entries })).into_response()));
}
if decision.route_kind.as_deref() == Some("aws_regions")
&& request_method == http::Method::GET
&& request_path == "/api/admin/system/aws-regions"
@@ -942,6 +1066,15 @@ fn query_param(query_string: Option<&str>, name: &str) -> Option<String> {
.find_map(|(key, value)| (key == name && !value.is_empty()).then(|| value.into_owned()))
}
fn query_flag(query_string: Option<&str>, name: &str) -> bool {
query_param(query_string, name).is_some_and(|value| {
matches!(
value.trim().to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
})
}
fn parse_older_than_days_query(query_string: Option<&str>) -> Result<Option<u32>, Response<Body>> {
let Some(value) = query_param(query_string, "older_than_days") else {
return Ok(None);

2
apps/aether-gateway/src/handlers/admin/system/shared/mod.rs
View File

@@ -4,3 +4,5 @@ pub(crate) mod modules;
pub(crate) mod paths;
pub(crate) mod settings;
pub(crate) mod smtp;
pub(crate) mod update;
pub(crate) mod update_client;

577
apps/aether-gateway/src/handlers/admin/system/shared/settings.rs
View File

@@ -1,11 +1,15 @@
use crate::handlers::admin::request::AdminAppState;
use crate::handlers::admin::shared::build_admin_usage_counter_health_payload;
use crate::handlers::admin::system::shared::update_client::{
build_direct_update_http_client, build_update_http_client, has_explicit_update_proxy_env,
update_github_token_from_env,
};
use crate::handlers::shared::{system_config_bool, system_config_string};
use crate::GatewayError;
use aether_admin::system::{
build_admin_api_formats_payload as build_admin_api_formats_payload_pure,
build_admin_system_check_update_payload as build_admin_system_check_update_payload_pure,
build_admin_system_check_update_payload_with_release,
build_admin_system_check_update_payload_with_release, build_admin_system_releases_payload,
build_admin_system_settings_payload as build_admin_system_settings_payload_pure,
build_admin_system_settings_updated_payload,
build_admin_system_stats_payload as build_admin_system_stats_payload_pure,
@@ -15,12 +19,18 @@ use axum::body::Bytes;
use axum::http;
#[cfg(not(test))]
use serde::Deserialize;
use serde_json::json;
#[cfg(not(test))]
use serde_json::{json, Value};
use std::time::Duration;
const AETHER_RELEASES_API_URL: &str =
"https://api.github.com/repos/fawney19/Aether/releases?per_page=20";
const AETHER_RELEASE_TAG_URL_BASE: &str = "https://github.com/fawney19/Aether/releases/tag";
const SOURCE_BUILD_UPDATE_BLOCKER: &str = "当前为源码构建，请使用 git pull 后重新编译。";
const SOURCE_BUILD_RELEASE_BLOCKER: &str = "当前为源码构建，请手动切换到对应标签后重新编译。";
/// Minimum interval between actual GitHub API requests. Within this window
/// the cached result is reused.
const RELEASE_CACHE_TTL: Duration = Duration::from_secs(1200);
pub(crate) fn current_aether_version() -> String {
option_env!("AETHER_BUILD_VERSION")
@@ -37,56 +47,388 @@ pub(crate) fn build_admin_system_check_update_payload_from_release(
latest_release: Option<AdminSystemUpdateRelease>,
error: Option<String>,
) -> serde_json::Value {
build_admin_system_check_update_payload_with_release(
let mut payload = build_admin_system_check_update_payload_with_release(
current_aether_version(),
latest_release,
error,
)
);
apply_source_build_check_update_override(&mut payload, current_build_is_release());
payload
}
pub(crate) fn build_admin_system_releases_list_payload(
releases: Vec<AdminSystemUpdateRelease>,
error: Option<String>,
) -> serde_json::Value {
let mut payload =
build_admin_system_releases_payload(current_aether_version(), releases, error);
apply_source_build_releases_override(&mut payload, current_build_is_release());
payload
}
fn current_build_is_release() -> bool {
option_env!("AETHER_BUILD_TYPE").unwrap_or("source") == "release"
}
fn apply_source_build_check_update_override(payload: &mut Value, release_build: bool) {
if release_build {
return;
}
if payload.get("has_update").and_then(Value::as_bool) != Some(true) {
return;
}
payload["updatable"] = json!(false);
payload["update_blocker"] = json!(SOURCE_BUILD_UPDATE_BLOCKER);
}
fn apply_source_build_releases_override(payload: &mut Value, release_build: bool) {
if release_build {
return;
}
let Some(releases) = payload.get_mut("releases").and_then(Value::as_array_mut) else {
return;
};
for release in releases {
if release.get("is_current").and_then(Value::as_bool) == Some(true) {
continue;
}
release["updatable"] = json!(false);
if release.get("update_blocker").is_none() || release["update_blocker"].is_null() {
release["update_blocker"] = json!(SOURCE_BUILD_RELEASE_BLOCKER);
}
}
}
#[cfg(not(test))]
struct CachedReleases {
all: Vec<AdminSystemUpdateRelease>,
latest: Option<AdminSystemUpdateRelease>,
error: Option<String>,
fetched_at: std::time::Instant,
}
#[cfg(not(test))]
fn releases_cache() -> &'static std::sync::Mutex<Option<CachedReleases>> {
static CACHE: std::sync::OnceLock<std::sync::Mutex<Option<CachedReleases>>> =
std::sync::OnceLock::new();
CACHE.get_or_init(|| std::sync::Mutex::new(None))
}
#[cfg(not(test))]
async fn ensure_releases_cached(force: bool) {
{
if let Ok(guard) = releases_cache().lock() {
if let Some(cached) = guard.as_ref() {
if should_reuse_releases_cache(
force,
cached.error.is_some(),
cached.fetched_at.elapsed(),
) {
return;
}
}
}
}
let (all, latest, error) = match fetch_admin_system_releases_inner().await {
Ok(releases) => {
let latest = releases.first().cloned();
(releases, latest, None)
}
Err(err) => (Vec::new(), None, Some(err)),
};
if let Ok(mut guard) = releases_cache().lock() {
*guard = Some(CachedReleases {
all,
latest,
error,
fetched_at: std::time::Instant::now(),
});
}
}
fn should_reuse_releases_cache(force: bool, has_error: bool, age: Duration) -> bool {
!force && !has_error && age < RELEASE_CACHE_TTL
}
#[cfg(not(test))]
pub(crate) async fn fetch_latest_admin_system_release(
force: bool,
) -> (Option<AdminSystemUpdateRelease>, Option<String>) {
match fetch_latest_admin_system_release_inner().await {
Ok(release) => (release, None),
Err(err) => (None, Some(err)),
ensure_releases_cached(force).await;
if let Ok(guard) = releases_cache().lock() {
if let Some(cached) = guard.as_ref() {
return (cached.latest.clone(), cached.error.clone());
}
}
(None, None)
}
#[cfg(not(test))]
pub(crate) async fn fetch_admin_system_releases(
force: bool,
) -> (Vec<AdminSystemUpdateRelease>, Option<String>) {
ensure_releases_cached(force).await;
if let Ok(guard) = releases_cache().lock() {
if let Some(cached) = guard.as_ref() {
return (cached.all.clone(), cached.error.clone());
}
}
(Vec::new(), None)
}
#[cfg(test)]
pub(crate) async fn fetch_admin_system_releases(
_force: bool,
) -> (Vec<AdminSystemUpdateRelease>, Option<String>) {
(
Vec::new(),
Some("测试环境未请求 GitHub Releases".to_string()),
)
}
#[cfg(test)]
pub(crate) async fn fetch_latest_admin_system_release(
_force: bool,
) -> (Option<AdminSystemUpdateRelease>, Option<String>) {
(None, Some("测试环境未请求 GitHub Releases".to_string()))
}
#[cfg(not(test))]
async fn fetch_latest_admin_system_release_inner(
) -> Result<Option<AdminSystemUpdateRelease>, String> {
let releases: Vec<GitHubRelease> = reqwest::Client::builder()
.timeout(Duration::from_secs(8))
.build()
.map_err(|err| format!("创建更新检查客户端失败: {err}"))?
.get(AETHER_RELEASES_API_URL)
.header(reqwest::header::USER_AGENT, "Aether-Gateway update-check")
.header(reqwest::header::ACCEPT, "application/vnd.github+json")
.send()
.await
.map_err(|err| format!("请求 GitHub Releases 失败: {err}"))?
.error_for_status()
.map_err(|err| format!("GitHub Releases 返回错误: {err}"))?
.json()
.await
.map_err(|err| format!("解析 GitHub Releases 失败: {err}"))?;
pub(crate) async fn resolve_update_target(
version: Option<String>,
) -> Result<(String, String, Option<String>), (http::StatusCode, serde_json::Value)> {
let (releases, error) = fetch_admin_system_releases(false).await;
if releases.is_empty() {
return Err((
http::StatusCode::SERVICE_UNAVAILABLE,
json!({ "detail": error.unwrap_or_else(|| "无法获取版本信息".to_string()) }),
));
}
let release = match version {
Some(ref v) => releases.into_iter().find(|r| r.version == *v),
None => releases.into_iter().next(),
};
let release = release.ok_or_else(|| {
(
http::StatusCode::NOT_FOUND,
json!({ "detail": "未找到指定版本" }),
)
})?;
let tarball_url = release.tarball_url.ok_or_else(|| {
(
http::StatusCode::PRECONDITION_REQUIRED,
json!({ "detail": format!("版本 {} 没有适用于当前平台的安装包", release.version) }),
)
})?;
let sha256sums_url = release.sha256sums_url.ok_or_else(|| {
(
http::StatusCode::PRECONDITION_REQUIRED,
json!({ "detail": format!("版本 {} 缺少 SHA256SUMS 校验文件，已拒绝在线更新", release.version) }),
)
})?;
Ok((release.version, tarball_url, Some(sha256sums_url)))
}
#[cfg(test)]
pub(crate) async fn resolve_update_target(
_version: Option<String>,
) -> Result<(String, String, Option<String>), (http::StatusCode, serde_json::Value)> {
Err((
http::StatusCode::SERVICE_UNAVAILABLE,
json!({ "detail": "测试环境不支持更新" }),
))
}
#[cfg(not(test))]
async fn fetch_admin_system_releases_inner() -> Result<Vec<AdminSystemUpdateRelease>, String> {
let current_channel = update_channel_for_version(&current_aether_version());
let timeout = Duration::from_secs(8);
let github_token = update_github_token_from_env();
let client = build_update_http_client(timeout, "更新检查")?;
let releases = match fetch_github_releases_with_client(&client, github_token.as_deref()).await {
Ok(releases) => releases,
Err(err) if err.rate_limited && !has_explicit_update_proxy_env() => {
let direct_client = build_direct_update_http_client(timeout, "更新检查直连重试")?;
fetch_github_releases_with_client(&direct_client, github_token.as_deref())
.await
.map_err(|retry_err| retry_err.message)?
}
Err(err) => return Err(err.message),
};
Ok(releases
.into_iter()
.find(|release| !release.draft && release.tag_name.starts_with('v'))
.map(|release| AdminSystemUpdateRelease {
version: release.tag_name,
release_url: Some(release.html_url),
release_notes: release.body.filter(|body| !body.trim().is_empty()),
published_at: release.published_at,
}))
.filter(|release| should_include_release_for_channel(release, current_channel))
.map(|release| {
let (tarball_url, sha256sums_url) = select_release_tarball_urls(&release);
AdminSystemUpdateRelease {
version: release.tag_name.clone(),
release_url: Some(github_release_tag_url(&release.tag_name)),
release_notes: release.body.filter(|body| !body.trim().is_empty()),
published_at: release.published_at,
tarball_url,
sha256sums_url,
}
})
.collect())
}
#[derive(Debug)]
struct GitHubReleaseFetchError {
message: String,
rate_limited: bool,
}
#[cfg(not(test))]
async fn fetch_github_releases_with_client(
client: &reqwest::Client,
github_token: Option<&str>,
) -> Result<Vec<GitHubRelease>, GitHubReleaseFetchError> {
let mut request = client
.get(AETHER_RELEASES_API_URL)
.header(reqwest::header::USER_AGENT, "Aether-Gateway update-check")
.header(reqwest::header::ACCEPT, "application/vnd.github+json");
if let Some(token) = github_token {
request = request.bearer_auth(token);
}
let response = request
.send()
.await
.map_err(|err| GitHubReleaseFetchError {
message: format!("请求 GitHub Releases 失败: {err}"),
rate_limited: false,
})?;
let status = response.status();
if !status.is_success() {
let body = response.text().await.unwrap_or_default();
return Err(github_release_response_error(status, &body));
}
response
.json()
.await
.map_err(|err| GitHubReleaseFetchError {
message: format!("解析 GitHub Releases 失败: {err}"),
rate_limited: false,
})
}
fn github_release_response_error(
status: reqwest::StatusCode,
response_body: &str,
) -> GitHubReleaseFetchError {
if is_github_rate_limit_error(status, response_body) {
return GitHubReleaseFetchError {
message: "GitHub Releases API 已触发限流；当前共享代理出口的匿名额度已用尽，更新检查将自动尝试直连。若仍失败，请配置 AETHER_UPDATE_GITHUB_TOKEN / GITHUB_TOKEN / GH_TOKEN，或为 GitHub 更新检查单独设置可用代理。".to_string(),
rate_limited: true,
};
}
let detail = parse_github_error_message(response_body);
let message = if detail.is_empty() {
format!("GitHub Releases 返回错误: HTTP {status}")
} else {
format!("GitHub Releases 返回错误: HTTP {status}; {detail}")
};
GitHubReleaseFetchError {
message,
rate_limited: false,
}
}
fn is_github_rate_limit_error(status: reqwest::StatusCode, response_body: &str) -> bool {
status == reqwest::StatusCode::FORBIDDEN
&& response_body
.to_ascii_lowercase()
.contains("rate limit exceeded")
}
fn parse_github_error_message(response_body: &str) -> String {
serde_json::from_str::<serde_json::Value>(response_body)
.ok()
.and_then(|value| {
value
.get("message")
.and_then(|message| message.as_str())
.map(str::trim)
.filter(|message| !message.is_empty())
.map(ToString::to_string)
})
.unwrap_or_default()
}
#[cfg(not(test))]
fn should_include_release_for_channel(
release: &GitHubRelease,
current_channel: UpdateChannel,
) -> bool {
if release.draft || !release.tag_name.starts_with('v') {
return false;
}
if !release.prerelease {
return true;
}
current_channel.allows_prerelease(&release.tag_name)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum UpdateChannel {
Stable,
Rc,
Beta,
OtherPrerelease,
}
impl UpdateChannel {
fn allows_prerelease(self, release_version: &str) -> bool {
match self {
Self::Stable => false,
Self::Rc => update_channel_for_version(release_version) == Self::Rc,
Self::Beta => update_channel_for_version(release_version) == Self::Beta,
Self::OtherPrerelease => {
matches!(
update_channel_for_version(release_version),
Self::Rc | Self::Beta | Self::OtherPrerelease
)
}
}
}
}
fn update_channel_for_version(version: &str) -> UpdateChannel {
let normalized = version
.trim()
.strip_prefix('v')
.or_else(|| version.trim().strip_prefix('V'))
.unwrap_or(version.trim());
let Some((_, prerelease)) = normalized.split_once('-') else {
return UpdateChannel::Stable;
};
let prerelease = prerelease.to_ascii_lowercase();
if prerelease.starts_with("rc") {
UpdateChannel::Rc
} else if prerelease.starts_with("beta") {
UpdateChannel::Beta
} else {
UpdateChannel::OtherPrerelease
}
}
#[cfg(not(test))]
#[derive(Debug, Deserialize)]
struct GitHubReleaseAsset {
name: String,
browser_download_url: String,
}
#[cfg(not(test))]
@@ -100,6 +442,43 @@ struct GitHubRelease {
published_at: Option<String>,
#[serde(default)]
draft: bool,
#[serde(default)]
prerelease: bool,
#[serde(default)]
assets: Vec<GitHubReleaseAsset>,
}
fn github_release_tag_url(tag_name: &str) -> String {
format!("{AETHER_RELEASE_TAG_URL_BASE}/{tag_name}")
}
#[cfg(not(test))]
fn select_release_tarball_urls(release: &GitHubRelease) -> (Option<String>, Option<String>) {
let platform = if cfg!(target_os = "macos") {
"macos"
} else {
"linux"
};
let arch = if cfg!(target_arch = "aarch64") {
"arm64"
} else {
"amd64"
};
let expected_name = format!("aether-{}-{}-{}.tar.gz", release.tag_name, platform, arch);
let tarball_url = release
.assets
.iter()
.find(|a| a.name == expected_name)
.map(|a| a.browser_download_url.clone());
let sha256sums_url = release
.assets
.iter()
.find(|a| a.name == "SHA256SUMS")
.map(|a| a.browser_download_url.clone());
(tarball_url, sha256sums_url)
}
pub(crate) async fn build_admin_system_stats_payload(
@@ -249,3 +628,137 @@ pub(crate) async fn apply_admin_system_settings_update(
pub(crate) fn build_admin_api_formats_payload() -> serde_json::Value {
build_admin_api_formats_payload_pure()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn update_channel_detects_stable_rc_beta_and_other_prerelease() {
assert_eq!(update_channel_for_version("v1.2.3"), UpdateChannel::Stable);
assert_eq!(update_channel_for_version("1.2.3-rc1"), UpdateChannel::Rc);
assert_eq!(
update_channel_for_version("1.2.3-beta.2"),
UpdateChannel::Beta
);
assert_eq!(
update_channel_for_version("1.2.3-alpha.1"),
UpdateChannel::OtherPrerelease
);
}
#[test]
fn stable_channel_does_not_allow_prereleases() {
assert!(!UpdateChannel::Stable.allows_prerelease("v1.2.3-rc1"));
}
#[test]
fn prerelease_channels_only_follow_matching_channel() {
assert!(UpdateChannel::Rc.allows_prerelease("v1.2.3-rc2"));
assert!(!UpdateChannel::Rc.allows_prerelease("v1.2.3-beta.1"));
assert!(UpdateChannel::Beta.allows_prerelease("v1.2.3-beta.2"));
assert!(!UpdateChannel::Beta.allows_prerelease("v1.2.3-rc1"));
assert!(UpdateChannel::OtherPrerelease.allows_prerelease("v1.2.3-alpha.1"));
assert!(UpdateChannel::OtherPrerelease.allows_prerelease("v1.2.3-rc1"));
}
#[test]
fn github_release_tag_url_points_to_explicit_tag_page() {
assert_eq!(
github_release_tag_url("v0.7.3"),
"https://github.com/fawney19/Aether/releases/tag/v0.7.3"
);
}
#[test]
fn successful_release_cache_is_reused_within_ttl() {
assert!(should_reuse_releases_cache(
false,
false,
Duration::from_secs(60)
));
}
#[test]
fn failed_release_cache_is_not_reused() {
assert!(!should_reuse_releases_cache(
false,
true,
Duration::from_secs(60)
));
}
#[test]
fn force_refresh_bypasses_release_cache() {
assert!(!should_reuse_releases_cache(
true,
false,
Duration::from_secs(60)
));
}
#[test]
fn github_rate_limit_response_is_marked_retryable() {
let err = github_release_response_error(
reqwest::StatusCode::FORBIDDEN,
r#"{"message":"API rate limit exceeded for 1.2.3.4."}"#,
);
assert!(err.rate_limited);
assert!(err.message.contains("GitHub Releases API 已触发限流"));
}
#[test]
fn non_rate_limit_github_response_keeps_http_status_message() {
let err = github_release_response_error(
reqwest::StatusCode::FORBIDDEN,
r#"{"message":"Resource not accessible"}"#,
);
assert!(!err.rate_limited);
assert!(err.message.contains("HTTP 403 Forbidden"));
assert!(err.message.contains("Resource not accessible"));
}
#[test]
fn source_build_check_update_override_marks_latest_release_non_updatable() {
let mut payload = json!({
"has_update": true,
"updatable": true,
"update_blocker": serde_json::Value::Null
});
apply_source_build_check_update_override(&mut payload, false);
assert_eq!(payload["updatable"], false);
assert_eq!(payload["update_blocker"], SOURCE_BUILD_UPDATE_BLOCKER);
}
#[test]
fn source_build_releases_override_marks_non_current_entries_non_updatable() {
let mut payload = json!({
"releases": [
{
"version": "v0.7.3",
"is_current": false,
"updatable": true,
"update_blocker": serde_json::Value::Null
},
{
"version": "v0.7.2",
"is_current": true,
"updatable": true,
"update_blocker": "当前版本"
}
]
});
apply_source_build_releases_override(&mut payload, false);
assert_eq!(payload["releases"][0]["updatable"], false);
assert_eq!(
payload["releases"][0]["update_blocker"],
SOURCE_BUILD_RELEASE_BLOCKER
);
assert_eq!(payload["releases"][1]["updatable"], true);
assert_eq!(payload["releases"][1]["update_blocker"], "当前版本");
}
}

998
apps/aether-gateway/src/handlers/admin/system/shared/update.rs Normal file
View File

@@ -0,0 +1,998 @@
use crate::handlers::admin::system::shared::update_client::build_update_http_client;
use crate::GatewayError;
use axum::http;
use futures_util::StreamExt;
use serde_json::json;
use sha2::{Digest, Sha256};
use std::path::Component;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Mutex;
#[derive(Debug, Clone, serde::Serialize)]
pub(crate) struct SystemUpdateTaskStatus {
pub phase: &'static str,
pub error: Option<String>,
pub output: Option<String>,
pub progress_label: Option<String>,
pub downloaded_bytes: Option<u64>,
pub total_bytes: Option<u64>,
pub progress_percent: Option<u8>,
}
static UPDATE_TASK_STATUS: std::sync::OnceLock<Mutex<SystemUpdateTaskStatus>> =
std::sync::OnceLock::new();
fn update_task_status_lock() -> &'static Mutex<SystemUpdateTaskStatus> {
UPDATE_TASK_STATUS.get_or_init(|| {
Mutex::new(SystemUpdateTaskStatus {
phase: "idle",
error: None,
output: None,
progress_label: None,
downloaded_bytes: None,
total_bytes: None,
progress_percent: None,
})
})
}
fn set_update_task_phase(phase: &'static str) {
if let Ok(mut guard) = update_task_status_lock().lock() {
guard.phase = phase;
guard.error = None;
guard.output = None;
guard.progress_label = None;
guard.downloaded_bytes = None;
guard.total_bytes = None;
guard.progress_percent = None;
}
}
fn set_update_task_download_progress(label: &str, downloaded_bytes: u64, total_bytes: Option<u64>) {
if let Ok(mut guard) = update_task_status_lock().lock() {
guard.progress_label = Some(label.to_string());
guard.downloaded_bytes = Some(downloaded_bytes);
guard.total_bytes = total_bytes;
guard.progress_percent = total_bytes
.filter(|total| *total > 0)
.map(|total| ((downloaded_bytes.saturating_mul(100) / total).min(100)) as u8);
}
}
fn set_update_task_failed(error: String) {
if let Ok(mut guard) = update_task_status_lock().lock() {
guard.phase = "failed";
guard.error = Some(error);
}
}
fn set_update_task_output(output: String) {
if let Ok(mut guard) = update_task_status_lock().lock() {
guard.output = Some(output);
}
}
pub(crate) fn read_update_task_status() -> SystemUpdateTaskStatus {
update_task_status_lock()
.lock()
.map(|guard| guard.clone())
.unwrap_or(SystemUpdateTaskStatus {
phase: "idle",
error: None,
output: None,
progress_label: None,
downloaded_bytes: None,
total_bytes: None,
progress_percent: None,
})
}
static PREPARED_VERSION: std::sync::OnceLock<Mutex<Option<String>>> = std::sync::OnceLock::new();
fn prepared_version_lock() -> &'static Mutex<Option<String>> {
PREPARED_VERSION.get_or_init(|| Mutex::new(None))
}
fn set_prepared_version(version: String) {
if let Ok(mut guard) = prepared_version_lock().lock() {
*guard = Some(version);
}
}
pub(crate) fn get_prepared_version() -> Option<String> {
prepared_version_lock().lock().ok()?.clone()
}
const UPDATE_HISTORY_FILENAME: &str = ".aether-update-history.json";
const PREVIOUS_RELEASE_FILENAME: &str = ".aether-previous-release";
const MAX_HISTORY_ENTRIES: usize = 50;
const RESTART_EXIT_CODE: i32 = 75;
const MAX_RELEASE_DOWNLOAD_BYTES: u64 = 512 * 1024 * 1024;
const MAX_SHA256SUMS_DOWNLOAD_BYTES: u64 = 1024 * 1024;
const MAX_EXTRACTED_RELEASE_BYTES: u64 = 1024 * 1024 * 1024;
const DEFAULT_UPDATE_DOWNLOAD_TIMEOUT_SECS: u64 = 600;
const DEFAULT_UPDATE_DOWNLOAD_IDLE_TIMEOUT_SECS: u64 = 30;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub(crate) struct UpdateHistoryEntry {
pub timestamp: String,
pub operation: String,
pub success: bool,
pub error: Option<String>,
pub output_tail: Option<String>,
}
fn aether_base_dir() -> PathBuf {
std::env::var("AETHER_BASE_DIR")
.ok()
.filter(|v| !v.is_empty())
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from("/opt/aether"))
}
fn releases_base_dir() -> PathBuf {
aether_base_dir().join("releases")
}
fn safe_release_name(version: &str) -> Result<String, String> {
let value = version.trim();
if value.is_empty() || value == "." || value == ".." {
return Err("版本号为空或非法".to_string());
}
if !value
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_' | '+'))
{
return Err(format!("版本号包含非法字符: {version}"));
}
Ok(value.to_string())
}
fn release_dir_for_version(version: &str) -> Result<PathBuf, String> {
Ok(releases_base_dir().join(safe_release_name(version)?))
}
fn current_symlink_path() -> PathBuf {
aether_base_dir().join("current")
}
fn current_release_name() -> Option<String> {
std::fs::read_link(current_symlink_path())
.ok()?
.file_name()
.and_then(|name| name.to_str())
.map(str::to_string)
}
fn update_history_path() -> PathBuf {
aether_base_dir().join(UPDATE_HISTORY_FILENAME)
}
fn append_update_history(
operation: &str,
success: bool,
error: Option<&str>,
output: Option<&str>,
) {
let path = update_history_path();
let entry = UpdateHistoryEntry {
timestamp: chrono::Utc::now().to_rfc3339(),
operation: operation.to_string(),
success,
error: error.map(|s| s.to_string()),
output_tail: output.map(|s| {
let lines: Vec<&str> = s.lines().collect();
let start = lines.len().saturating_sub(20);
lines[start..].join("\n")
}),
};
let mut entries: Vec<UpdateHistoryEntry> = std::fs::read_to_string(&path)
.ok()
.and_then(|content| serde_json::from_str(&content).ok())
.unwrap_or_default();
entries.push(entry);
if entries.len() > MAX_HISTORY_ENTRIES {
entries.drain(..entries.len() - MAX_HISTORY_ENTRIES);
}
if let Ok(json) = serde_json::to_string_pretty(&entries) {
let _ = std::fs::write(&path, json);
}
}
pub(crate) fn read_update_history() -> Vec<UpdateHistoryEntry> {
let path = update_history_path();
std::fs::read_to_string(&path)
.ok()
.and_then(|content| serde_json::from_str(&content).ok())
.unwrap_or_default()
}
static SYSTEM_UPDATE_RUNNING: AtomicBool = AtomicBool::new(false);
struct SystemUpdateGuard;
impl SystemUpdateGuard {
fn try_acquire() -> Option<Self> {
if SYSTEM_UPDATE_RUNNING
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
.is_ok()
{
Some(Self)
} else {
None
}
}
}
impl Drop for SystemUpdateGuard {
fn drop(&mut self) {
SYSTEM_UPDATE_RUNNING.store(false, Ordering::SeqCst);
}
}
fn current_build_type() -> &'static str {
option_env!("AETHER_BUILD_TYPE").unwrap_or("source")
}
#[cfg(not(test))]
fn is_release_build() -> bool {
current_build_type() == "release"
}
#[cfg(test)]
fn is_release_build() -> bool {
true
}
pub(crate) fn build_admin_system_update_capability_payload() -> serde_json::Value {
let supported = is_release_build();
let build_type = current_build_type();
let rollback_available = find_rollback_target().is_some();
let task_status = read_update_task_status();
let base_dir = aether_base_dir();
json!({
"supported": supported,
"enabled": supported,
"rollback_available": rollback_available,
"task_status": task_status.phase,
"task_error": task_status.error,
"build_type": build_type,
"install_root": base_dir,
"message": if supported {
"一键更新可用"
} else {
"源码构建不支持在线更新"
},
})
}
fn find_rollback_target() -> Option<String> {
let previous_path = aether_base_dir().join(PREVIOUS_RELEASE_FILENAME);
let previous = std::fs::read_to_string(previous_path).ok()?;
let previous = previous.trim().to_string();
if previous.is_empty() {
return None;
}
let target_dir = release_dir_for_version(&previous).ok()?;
if target_dir.is_dir() {
Some(previous)
} else {
None
}
}
pub(crate) async fn prepare_admin_system_update_task(
version: String,
tarball_url: String,
sha256sums_url: Option<String>,
) -> Result<Result<serde_json::Value, (http::StatusCode, serde_json::Value)>, GatewayError> {
if !is_release_build() {
return Ok(Err(source_build_rejection_response()));
}
let Some(sha256sums_url) = sha256sums_url.filter(|url| !url.trim().is_empty()) else {
return Ok(Err((
http::StatusCode::PRECONDITION_REQUIRED,
json!({ "detail": "缺少 SHA256SUMS 校验文件，已拒绝在线更新" }),
)));
};
let Some(guard) = SystemUpdateGuard::try_acquire() else {
return Ok(Err(update_already_running_response()));
};
set_update_task_phase("preparing");
tokio::spawn(async move {
let _guard = guard;
let total_timeout = update_download_total_timeout();
let result = match tokio::time::timeout(
total_timeout,
download_and_extract_release(&version, &tarball_url, &sha256sums_url),
)
.await
{
Ok(result) => result,
Err(_) => Err(format!(
"下载更新包超时: 超过 {} 秒",
total_timeout.as_secs()
)),
};
match result {
Ok(output) => {
set_update_task_phase("prepared");
set_update_task_output(output.clone());
set_prepared_version(version);
append_update_history("prepare", true, None, Some(&output));
}
Err(err) => {
set_update_task_failed(err.clone());
append_update_history("prepare", false, Some(&err), None);
}
}
});
Ok(Ok(json!({
"message": "更新包开始下载，请等待准备完成",
"started": true,
"need_restart": false,
})))
}
async fn download_and_extract_release(
version: &str,
tarball_url: &str,
sha256sums_url: &str,
) -> Result<String, String> {
let client = build_update_http_client(update_download_total_timeout(), "更新下载")?;
set_update_task_phase("downloading");
let tarball_bytes =
download_update_bytes(&client, tarball_url, MAX_RELEASE_DOWNLOAD_BYTES, "更新包").await?;
set_update_task_phase("downloading_checksum");
let sha256_text = String::from_utf8(
download_update_bytes(
&client,
sha256sums_url,
MAX_SHA256SUMS_DOWNLOAD_BYTES,
"校验文件",
)
.await?,
)
.map_err(|err| format!("校验文件不是有效 UTF-8: {err}"))?;
let tarball_url_owned = tarball_url.to_string();
let version_owned = version.to_string();
tokio::task::spawn_blocking(move || {
set_update_task_phase("verifying");
verify_sha256(&tarball_bytes, &sha256_text, &tarball_url_owned)?;
set_update_task_phase("extracting");
extract_release(&version_owned, &tarball_bytes)
})
.await
.map_err(|err| format!("\u{89e3}\u{538b}\u{4efb}\u{52a1}\u{5f02}\u{5e38}: {err}"))?
}
async fn download_update_bytes(
client: &reqwest::Client,
url: &str,
max_bytes: u64,
label: &str,
) -> Result<Vec<u8>, String> {
validate_update_download_url(url)?;
let idle_timeout = update_download_idle_timeout();
let response = tokio::time::timeout(
idle_timeout,
client
.get(url)
.header(reqwest::header::USER_AGENT, "Aether-Gateway update")
.send(),
)
.await
.map_err(|_| {
format!(
"下载{label}超时: {} 秒内没有收到响应",
idle_timeout.as_secs()
)
})?
.map_err(|err| format!("下载{label}失败: {err}"))?
.error_for_status()
.map_err(|err| format!("下载{label}返回错误: {err}"))?;
if let Some(content_length) = response.content_length() {
if content_length > max_bytes {
return Err(format!(
"{label}过大: {content_length} bytes，最大允许 {max_bytes} bytes"
));
}
}
let total_bytes = response.content_length();
let mut stream = response.bytes_stream();
let mut data = Vec::new();
set_update_task_download_progress(label, 0, total_bytes);
while let Some(chunk) = tokio::time::timeout(idle_timeout, stream.next())
.await
.map_err(|_| {
format!(
"下载{label}超时: {} 秒内没有收到数据",
idle_timeout.as_secs()
)
})?
{
let chunk = chunk.map_err(|err| format!("读取{label}数据失败: {err}"))?;
let next_len = data.len() as u64 + chunk.len() as u64;
if next_len > max_bytes {
return Err(format!("{label}超过大小限制: 最大允许 {max_bytes} bytes"));
}
data.extend_from_slice(&chunk);
set_update_task_download_progress(label, data.len() as u64, total_bytes);
}
Ok(data)
}
fn update_download_total_timeout() -> std::time::Duration {
update_timeout_from_env(
"AETHER_UPDATE_DOWNLOAD_TIMEOUT_SECS",
DEFAULT_UPDATE_DOWNLOAD_TIMEOUT_SECS,
)
}
fn update_download_idle_timeout() -> std::time::Duration {
update_timeout_from_env(
"AETHER_UPDATE_DOWNLOAD_IDLE_TIMEOUT_SECS",
DEFAULT_UPDATE_DOWNLOAD_IDLE_TIMEOUT_SECS,
)
}
fn update_timeout_from_env(key: &str, default_secs: u64) -> std::time::Duration {
let secs = std::env::var(key)
.ok()
.and_then(|value| value.trim().parse::<u64>().ok())
.filter(|value| *value > 0)
.unwrap_or(default_secs);
std::time::Duration::from_secs(secs)
}
fn validate_update_download_url(raw_url: &str) -> Result<(), String> {
let parsed = url::Url::parse(raw_url).map_err(|err| format!("下载 URL 无效: {err}"))?;
if parsed.scheme() != "https" {
return Err("下载 URL 必须使用 HTTPS".to_string());
}
let Some(host) = parsed.host_str() else {
return Err("下载 URL 缺少主机名".to_string());
};
if host == "github.com"
|| host.ends_with(".github.com")
|| host == "objects.githubusercontent.com"
|| host.ends_with(".objects.githubusercontent.com")
{
return Ok(());
}
Err(format!("下载 URL 主机不受信任: {host}"))
}
fn verify_sha256(data: &[u8], sums_text: &str, tarball_url: &str) -> Result<(), String> {
let tarball_filename = tarball_url.rsplit('/').next().ok_or_else(|| {
"\u{65e0}\u{6cd5}\u{4ece} URL \u{63d0}\u{53d6}\u{6587}\u{4ef6}\u{540d}".to_string()
})?;
let expected_hash = sums_text
.lines()
.find_map(|line| {
let (hash, name) = line.split_once(char::is_whitespace)?;
let name = name.trim().trim_start_matches('*');
if name == tarball_filename {
Some(hash.to_string())
} else {
None
}
})
.ok_or_else(|| {
format!("SHA256SUMS \u{4e2d}\u{672a}\u{627e}\u{5230} {tarball_filename} \u{7684}\u{6821}\u{9a8c}\u{503c}")
})?;
let mut hasher = Sha256::new();
hasher.update(data);
let hash = hasher.finalize();
let actual_hash: String = hash.iter().map(|b| format!("{b:02x}")).collect();
if actual_hash != expected_hash {
return Err(format!(
"SHA256 \u{6821}\u{9a8c}\u{5931}\u{8d25}: \u{671f}\u{671b} {expected_hash}, \u{5b9e}\u{9645} {actual_hash}"
));
}
Ok(())
}
fn extract_release(version: &str, tarball_bytes: &[u8]) -> Result<String, String> {
let safe_version = safe_release_name(version)?;
if current_release_name().as_deref() == Some(safe_version.as_str()) {
return Err(format!("版本 {version} 已经是当前运行版本"));
}
let base_dir = releases_base_dir();
std::fs::create_dir_all(&base_dir).map_err(|err| {
format!("\u{521b}\u{5efa} releases \u{76ee}\u{5f55}\u{5931}\u{8d25}: {err}")
})?;
let release_dir = base_dir.join(&safe_version);
let staging_dir = base_dir.join(format!(".prepare-{}-{}", safe_version, std::process::id()));
remove_path_if_exists(&staging_dir).map_err(|err| format!("清理临时版本目录失败: {err}"))?;
std::fs::create_dir_all(&staging_dir).map_err(|err| format!("创建临时版本目录失败: {err}"))?;
if let Err(err) = unpack_release_archive(tarball_bytes, &staging_dir) {
let _ = remove_path_if_exists(&staging_dir);
return Err(err);
}
let bundle_dir = match find_release_payload_dir(&staging_dir) {
Ok(dir) => dir,
Err(err) => {
let _ = remove_path_if_exists(&staging_dir);
return Err(err);
}
};
if let Err(err) = validate_release_payload_dir(&bundle_dir) {
let _ = remove_path_if_exists(&staging_dir);
return Err(err);
}
remove_path_if_exists(&release_dir).map_err(|err| {
format!("\u{6e05}\u{7406}\u{65e7}\u{7248}\u{672c}\u{76ee}\u{5f55}\u{5931}\u{8d25}: {err}")
})?;
if bundle_dir == staging_dir {
std::fs::rename(&staging_dir, &release_dir)
.map_err(|err| format!("安装版本目录失败: {err}"))?;
} else {
std::fs::rename(&bundle_dir, &release_dir)
.map_err(|err| format!("安装版本目录失败: {err}"))?;
let _ = remove_path_if_exists(&staging_dir);
}
ensure_release_binary_permissions(&release_dir.join("bin/aether-gateway"));
Ok(format!(
"\u{7248}\u{672c} {} \u{5df2}\u{51c6}\u{5907}\u{5c31}\u{7eea}",
version
))
}
fn unpack_release_archive(tarball_bytes: &[u8], staging_dir: &Path) -> Result<(), String> {
let decoder = flate2::read::GzDecoder::new(std::io::Cursor::new(tarball_bytes));
let mut archive = tar::Archive::new(decoder);
let entries = archive
.entries()
.map_err(|err| format!("读取更新包失败: {err}"))?;
let mut extracted_bytes = 0u64;
for entry in entries {
let mut entry = entry.map_err(|err| format!("读取更新包条目失败: {err}"))?;
let path = entry
.path()
.map_err(|err| format!("读取更新包路径失败: {err}"))?
.to_path_buf();
validate_archive_entry_path(&path)?;
let entry_type = entry.header().entry_type();
if entry_type.is_file() {
let size = entry
.header()
.size()
.map_err(|err| format!("读取更新包文件大小失败: {err}"))?;
extracted_bytes = extracted_bytes.saturating_add(size);
if extracted_bytes > MAX_EXTRACTED_RELEASE_BYTES {
return Err(format!(
"更新包解压后过大: 最大允许 {MAX_EXTRACTED_RELEASE_BYTES} bytes"
));
}
} else if !entry_type.is_dir() {
return Err(format!("更新包包含不支持的条目: {}", path.display()));
}
let unpacked = entry
.unpack_in(staging_dir)
.map_err(|err| format!("解压更新包失败: {err}"))?;
if !unpacked {
return Err(format!("更新包包含非法路径: {}", path.display()));
}
}
Ok(())
}
fn validate_archive_entry_path(path: &Path) -> Result<(), String> {
let mut has_normal_component = false;
for component in path.components() {
match component {
Component::Normal(_) => has_normal_component = true,
Component::CurDir => {}
Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
return Err(format!("更新包包含非法路径: {}", path.display()));
}
}
}
if has_normal_component {
Ok(())
} else {
Err("更新包包含空路径".to_string())
}
}
fn find_release_payload_dir(staging_dir: &Path) -> Result<PathBuf, String> {
if looks_like_release_payload(staging_dir) {
return Ok(staging_dir.to_path_buf());
}
let mut candidates = Vec::new();
let entries =
std::fs::read_dir(staging_dir).map_err(|err| format!("读取更新包目录失败: {err}"))?;
for entry in entries {
let entry = entry.map_err(|err| format!("读取更新包条目失败: {err}"))?;
let path = entry.path();
if path.is_dir() && looks_like_release_payload(&path) {
candidates.push(path);
}
}
match candidates.len() {
1 => Ok(candidates.remove(0)),
0 => Err(
"\u{66f4}\u{65b0}\u{5305}\u{4e2d}\u{672a}\u{627e}\u{5230} bin/aether-gateway"
.to_string(),
),
_ => Err("更新包中包含多个可安装目录，无法确定目标版本".to_string()),
}
}
fn looks_like_release_payload(path: &Path) -> bool {
path.join("bin/aether-gateway").is_file() && path.join("frontend").is_dir()
}
fn validate_release_payload_dir(path: &Path) -> Result<(), String> {
if !path.join("bin/aether-gateway").is_file() {
return Err(
"\u{66f4}\u{65b0}\u{5305}\u{4e2d}\u{672a}\u{627e}\u{5230} bin/aether-gateway"
.to_string(),
);
}
if !path.join("frontend/index.html").is_file() {
return Err("更新包中未找到 frontend/index.html".to_string());
}
Ok(())
}
fn ensure_release_binary_permissions(binary_path: &Path) {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let _ = std::fs::set_permissions(binary_path, std::fs::Permissions::from_mode(0o755));
}
}
fn remove_path_if_exists(path: &Path) -> std::io::Result<()> {
match std::fs::symlink_metadata(path) {
Ok(meta) if meta.is_dir() && !meta.file_type().is_symlink() => {
std::fs::remove_dir_all(path)
}
Ok(_) => std::fs::remove_file(path),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(err),
}
}
pub(crate) async fn start_admin_system_update_task(
version: Option<String>,
) -> Result<Result<serde_json::Value, (http::StatusCode, serde_json::Value)>, GatewayError> {
if !is_release_build() {
return Ok(Err(source_build_rejection_response()));
}
let version = match version.or_else(get_prepared_version) {
Some(v) => v,
None => {
return Ok(Err((
http::StatusCode::BAD_REQUEST,
json!({ "detail": "\u{672a}\u{6307}\u{5b9a}\u{7248}\u{672c}\u{4e14}\u{6ca1}\u{6709}\u{5df2}\u{51c6}\u{5907}\u{7684}\u{66f4}\u{65b0}" }),
)));
}
};
let release_dir = match release_dir_for_version(&version) {
Ok(dir) => dir,
Err(err) => {
return Ok(Err((
http::StatusCode::BAD_REQUEST,
json!({ "detail": err }),
)));
}
};
if !release_dir.join("bin/aether-gateway").is_file() {
return Ok(Err((
http::StatusCode::PRECONDITION_REQUIRED,
json!({ "detail": format!("\u{7248}\u{672c} {version} \u{5c1a}\u{672a}\u{51c6}\u{5907}\u{597d}\u{ff0c}\u{8bf7}\u{5148}\u{6267}\u{884c} prepare-update") }),
)));
}
let Some(guard) = SystemUpdateGuard::try_acquire() else {
return Ok(Err(update_already_running_response()));
};
save_previous_release();
set_update_task_phase("restarting");
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
match switch_current_symlink(&version) {
Ok(_) => {
append_update_history(
"apply",
true,
None,
Some(&format!(
"\u{5df2}\u{5207}\u{6362}\u{5230}\u{7248}\u{672c} {version}"
)),
);
tracing::info!(version = %version, "update applied, exiting for restart");
request_process_restart();
}
Err(err) => {
tracing::error!(error = %err, "admin system update apply failed");
append_update_history("apply", false, Some(&err), None);
set_update_task_failed(err);
}
}
drop(guard);
});
Ok(Ok(json!({
"message": "\u{6b63}\u{5728}\u{5207}\u{6362}\u{7248}\u{672c}\u{5e76}\u{91cd}\u{542f}\u{ff0c}\u{670d}\u{52a1}\u{4f1a}\u{77ed}\u{6682}\u{4e0d}\u{53ef}\u{7528}",
"started": true,
"need_restart": true,
})))
}
fn save_previous_release() {
let current = current_symlink_path();
if let Ok(target) = std::fs::read_link(&current) {
if let Some(name) = target.file_name().and_then(|n| n.to_str()) {
let prev_path = aether_base_dir().join(PREVIOUS_RELEASE_FILENAME);
let _ = std::fs::write(prev_path, name);
}
}
}
fn switch_current_symlink(version: &str) -> Result<(), String> {
let target = release_dir_for_version(version)?;
if !target.is_dir() {
return Err(format!(
"\u{7248}\u{672c}\u{76ee}\u{5f55}\u{4e0d}\u{5b58}\u{5728}: {}",
target.display()
));
}
validate_release_payload_dir(&target)?;
let current = current_symlink_path();
let current_new = current.with_file_name("current.new");
let _ = remove_path_if_exists(&current_new);
#[cfg(unix)]
std::os::unix::fs::symlink(&target, &current_new)
.map_err(|err| format!("\u{521b}\u{5efa}\u{4e34}\u{65f6}\u{7b26}\u{53f7}\u{94fe}\u{63a5}\u{5931}\u{8d25}: {err}"))?;
#[cfg(windows)]
std::os::windows::fs::symlink_dir(&target, &current_new)
.map_err(|err| format!("\u{521b}\u{5efa}\u{4e34}\u{65f6}\u{7b26}\u{53f7}\u{94fe}\u{63a5}\u{5931}\u{8d25}: {err}"))?;
std::fs::rename(&current_new, &current)
.map_err(|err| format!("\u{539f}\u{5b50}\u{5207}\u{6362}\u{7b26}\u{53f7}\u{94fe}\u{63a5}\u{5931}\u{8d25}: {err}"))?;
Ok(())
}
pub(crate) async fn start_admin_system_rollback_task(
) -> Result<Result<serde_json::Value, (http::StatusCode, serde_json::Value)>, GatewayError> {
if !is_release_build() {
return Ok(Err(source_build_rejection_response()));
}
let Some(previous) = find_rollback_target() else {
return Ok(Err((
http::StatusCode::PRECONDITION_REQUIRED,
json!({ "detail": "\u{6ca1}\u{6709}\u{53ef}\u{56de}\u{6eda}\u{7684}\u{7248}\u{672c}" }),
)));
};
let Some(guard) = SystemUpdateGuard::try_acquire() else {
return Ok(Err(update_already_running_response()));
};
set_update_task_phase("rolling_back");
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
match switch_current_symlink(&previous) {
Ok(_) => {
let prev_path = aether_base_dir().join(PREVIOUS_RELEASE_FILENAME);
let _ = std::fs::remove_file(prev_path);
append_update_history(
"rollback",
true,
None,
Some(&format!(
"\u{5df2}\u{56de}\u{6eda}\u{5230}\u{7248}\u{672c} {previous}"
)),
);
tracing::info!(version = %previous, "rollback applied, exiting for restart");
request_process_restart();
}
Err(err) => {
tracing::error!(error = %err, "admin system rollback failed");
append_update_history("rollback", false, Some(&err), None);
set_update_task_failed(err);
}
}
drop(guard);
});
Ok(Ok(json!({
"message": "\u{56de}\u{6eda}\u{5df2}\u{542f}\u{52a8}\u{ff0c}\u{670d}\u{52a1}\u{4f1a}\u{77ed}\u{6682}\u{4e0d}\u{53ef}\u{7528}",
"started": true,
"need_restart": true,
})))
}
fn request_process_restart() -> ! {
std::process::exit(RESTART_EXIT_CODE);
}
fn update_already_running_response() -> (http::StatusCode, serde_json::Value) {
(
http::StatusCode::CONFLICT,
json!({ "detail": "\u{5df2}\u{6709}\u{4e00}\u{952e}\u{66f4}\u{65b0}\u{4efb}\u{52a1}\u{6b63}\u{5728}\u{6267}\u{884c}" }),
)
}
fn source_build_rejection_response() -> (http::StatusCode, serde_json::Value) {
(
http::StatusCode::PRECONDITION_REQUIRED,
json!({ "detail": "\u{6e90}\u{7801}\u{6784}\u{5efa}\u{4e0d}\u{652f}\u{6301}\u{5728}\u{7ebf}\u{66f4}\u{65b0}\u{ff0c}\u{8bf7}\u{4f7f}\u{7528}\u{6b63}\u{5f0f}\u{53d1}\u{5e03}\u{7248}" }),
)
}
#[cfg(test)]
mod tests {
use super::*;
use flate2::Compression;
fn temp_test_dir(name: &str) -> PathBuf {
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock should be after epoch")
.as_nanos();
std::env::temp_dir().join(format!(
"aether-update-{name}-{}-{nanos}",
std::process::id()
))
}
fn write_release_payload(root: &Path) {
std::fs::create_dir_all(root.join("bin")).expect("bin dir should be created");
std::fs::create_dir_all(root.join("frontend")).expect("frontend dir should be created");
std::fs::write(root.join("bin/aether-gateway"), b"test-binary")
.expect("binary should be written");
std::fs::write(root.join("frontend/index.html"), b"<html></html>")
.expect("frontend index should be written");
}
#[test]
fn update_finds_nested_release_payload_dir() {
let staging = temp_test_dir("nested");
let bundle = staging.join("aether-v1.2.3-linux-amd64");
write_release_payload(&bundle);
let found = find_release_payload_dir(&staging).expect("payload dir should be found");
assert_eq!(found, bundle);
std::fs::remove_dir_all(staging).ok();
}
#[test]
fn update_finds_flat_release_payload_dir() {
let staging = temp_test_dir("flat");
write_release_payload(&staging);
let found = find_release_payload_dir(&staging).expect("payload dir should be found");
assert_eq!(found, staging);
std::fs::remove_dir_all(found).ok();
}
#[test]
fn update_rejects_unsafe_release_names() {
assert!(safe_release_name("v1.2.3").is_ok());
assert!(safe_release_name("../v1.2.3").is_err());
assert!(safe_release_name("v1.2.3/linux").is_err());
assert!(safe_release_name("").is_err());
}
#[test]
fn update_validates_download_urls() {
assert!(validate_update_download_url(
"https://github.com/fawney19/Aether/releases/download/v1/aether.tar.gz"
)
.is_ok());
assert!(validate_update_download_url(
"https://objects.githubusercontent.com/github-production-release-asset/test"
)
.is_ok());
assert!(validate_update_download_url(
"http://github.com/fawney19/Aether/releases/download/v1/aether.tar.gz"
)
.is_err());
assert!(validate_update_download_url("https://example.com/aether.tar.gz").is_err());
}
#[test]
fn update_rejects_archive_path_traversal() {
assert!(validate_archive_entry_path(Path::new("bundle/bin/aether-gateway")).is_ok());
assert!(validate_archive_entry_path(Path::new("../escape")).is_err());
assert!(validate_archive_entry_path(Path::new("/tmp/escape")).is_err());
}
#[test]
fn update_rejects_archive_symlinks() {
let staging = temp_test_dir("symlink");
std::fs::create_dir_all(&staging).expect("staging dir should be created");
let mut encoder = flate2::write::GzEncoder::new(Vec::new(), Compression::default());
{
let mut builder = tar::Builder::new(&mut encoder);
let mut header = tar::Header::new_gnu();
header.set_entry_type(tar::EntryType::Symlink);
header.set_size(0);
header.set_mode(0o777);
header
.set_link_name("/bin/sh")
.expect("link name should be set");
header.set_cksum();
builder
.append_data(&mut header, "bundle/bin/aether-gateway", std::io::empty())
.expect("symlink entry should be appended");
builder.finish().expect("tar builder should finish");
}
let tarball = encoder.finish().expect("gzip encoder should finish");
let err = unpack_release_archive(&tarball, &staging).expect_err("archive should fail");
assert!(err.contains("不支持的条目"));
std::fs::remove_dir_all(staging).ok();
}
#[test]
fn update_verifies_sha256sum_for_asset_name() {
let data = b"release-bytes";
let mut hasher = Sha256::new();
hasher.update(data);
let expected: String = hasher
.finalize()
.iter()
.map(|b| format!("{b:02x}"))
.collect();
let sums = format!("{expected} aether-v1.2.3-linux-amd64.tar.gz\n");
verify_sha256(
data,
&sums,
"https://example.test/aether-v1.2.3-linux-amd64.tar.gz",
)
.expect("sha256 should match");
}
}

68
apps/aether-gateway/src/handlers/admin/system/shared/update_client.rs Normal file
View File

@@ -0,0 +1,68 @@
use std::time::Duration;
const EXPLICIT_UPDATE_PROXY_ENV_KEYS: &[&str] = &["AETHER_UPDATE_PROXY_URL", "UPDATE_PROXY_URL"];
const UPDATE_PROXY_ENV_KEYS: &[&str] = &[
"AETHER_UPDATE_PROXY_URL",
"UPDATE_PROXY_URL",
"HTTPS_PROXY",
"https_proxy",
"ALL_PROXY",
"all_proxy",
"HTTP_PROXY",
"http_proxy",
];
const UPDATE_GITHUB_TOKEN_ENV_KEYS: &[&str] =
&["AETHER_UPDATE_GITHUB_TOKEN", "GITHUB_TOKEN", "GH_TOKEN"];
pub(crate) fn build_update_http_client(
timeout: Duration,
label: &str,
) -> Result<reqwest::Client, String> {
let mut builder = base_update_http_client_builder(timeout);
if let Some(proxy_url) = update_proxy_url_from_env() {
let proxy = reqwest::Proxy::all(proxy_url)
.map_err(|_| format!("创建{label}代理失败，请检查更新代理环境变量"))?
.no_proxy(reqwest::NoProxy::from_env());
builder = builder.proxy(proxy);
}
builder
.build()
.map_err(|err| format!("创建{label}客户端失败: {err}"))
}
pub(crate) fn build_direct_update_http_client(
timeout: Duration,
label: &str,
) -> Result<reqwest::Client, String> {
base_update_http_client_builder(timeout)
.no_proxy()
.build()
.map_err(|err| format!("创建{label}客户端失败: {err}"))
}
pub(crate) fn has_explicit_update_proxy_env() -> bool {
read_nonempty_env_value(EXPLICIT_UPDATE_PROXY_ENV_KEYS).is_some()
}
fn base_update_http_client_builder(timeout: Duration) -> reqwest::ClientBuilder {
reqwest::Client::builder().timeout(timeout)
}
fn update_proxy_url_from_env() -> Option<String> {
read_nonempty_env_value(UPDATE_PROXY_ENV_KEYS)
}
pub(crate) fn update_github_token_from_env() -> Option<String> {
read_nonempty_env_value(UPDATE_GITHUB_TOKEN_ENV_KEYS)
}
fn read_nonempty_env_value(keys: &[&str]) -> Option<String> {
keys.iter().find_map(|key| {
std::env::var(key)
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
})
}

14
apps/aether-gateway/src/router.rs
View File

@@ -1,7 +1,8 @@
use std::path::PathBuf;
use axum::extract::Request;
use axum::http::Method;
use axum::http::header::{CACHE_CONTROL, EXPIRES, PRAGMA};
use axum::http::{HeaderValue, Method};
use axum::response::{IntoResponse, Response};
use axum::routing::any;
use axum::Router;
@@ -111,7 +112,16 @@ async fn serve_static_asset(static_dir: &PathBuf, request: Request) -> Response
async fn serve_frontend_index(index_html: &PathBuf, request: Request) -> Response {
match ServeFile::new(index_html).oneshot(request).await {
Ok(response) => response.into_response(),
Ok(mut response) => {
let headers = response.headers_mut();
headers.insert(
CACHE_CONTROL,
HeaderValue::from_static("no-store, no-cache, must-revalidate"),
);
headers.insert(PRAGMA, HeaderValue::from_static("no-cache"));
headers.insert(EXPIRES, HeaderValue::from_static("0"));
response.into_response()
}
Err(err) => {
warn!(error = %err, "failed to serve frontend index");
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()

265
apps/aether-gateway/src/tests/control/admin/system.rs
View File

@@ -36,6 +36,8 @@ use crate::constants::{
};
use crate::data::GatewayDataState;
static SYSTEM_UPDATE_TEST_MUTEX: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
#[tokio::test]
async fn gateway_handles_admin_system_version_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));
@@ -158,6 +160,269 @@ async fn gateway_handles_admin_system_check_update_locally_with_bearer_admin_ses
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_system_update_capability_locally() {
let _lock = SYSTEM_UPDATE_TEST_MUTEX.lock().await;
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/system/update-capability",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!("{gateway_url}/api/admin/system/update-capability"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["supported"].is_boolean());
assert!(payload["build_type"].is_string());
assert!(payload["task_status"].is_string());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_prepares_admin_system_update_locally() {
let _lock = SYSTEM_UPDATE_TEST_MUTEX.lock().await;
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/system/prepare-update",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/api/admin/system/prepare-update"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["detail"].is_string());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_admin_system_apply_update_without_prepared_version() {
let _lock = SYSTEM_UPDATE_TEST_MUTEX.lock().await;
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/system/apply-update",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/api/admin/system/apply-update"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["detail"].is_string());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_admin_system_rollback_without_previous_release() {
let _lock = SYSTEM_UPDATE_TEST_MUTEX.lock().await;
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/system/rollback",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/api/admin/system/rollback"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::PRECONDITION_REQUIRED);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["detail"].is_string());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_system_releases_locally() {
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/system/releases",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!("{gateway_url}/api/admin/system/releases"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["current_version"]
.as_str()
.is_some_and(|value| !value.is_empty()));
assert!(payload["releases"].is_array());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_admin_system_apply_update_with_nonexistent_version() {
let _lock = SYSTEM_UPDATE_TEST_MUTEX.lock().await;
let upstream_hits = Arc::new(Mutex::new(0usize));
let upstream_hits_clone = Arc::clone(&upstream_hits);
let upstream = Router::new().route(
"/api/admin/system/apply-update",
any(move |_request: Request| {
let upstream_hits_inner = Arc::clone(&upstream_hits_clone);
async move {
*upstream_hits_inner.lock().expect("mutex should lock") += 1;
(StatusCode::OK, Body::from("unexpected upstream hit"))
}
}),
);
let (upstream_url, upstream_handle) = start_server(upstream).await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!("{gateway_url}/api/admin/system/apply-update"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.header("content-type", "application/json")
.body(r#"{"version":"v99.99.99"}"#)
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["detail"].is_string());
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
gateway_handle.abort();
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_system_update_status_locally() {
let _lock = SYSTEM_UPDATE_TEST_MUTEX.lock().await;
let gateway = build_router_with_state(AppState::new().expect("gateway should build"));
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.get(format!("{gateway_url}/api/admin/system/update-status"))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.send()
.await
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert!(payload["phase"].is_string());
gateway_handle.abort();
}
#[tokio::test]
async fn gateway_handles_admin_system_aws_regions_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));

14
apps/aether-gateway/src/tests/frontdoor/core.rs
View File

@@ -145,6 +145,13 @@ async fn gateway_serves_frontend_routes_and_assets_without_shadowing_public_api(
.expect("request should succeed");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("cache-control")
.and_then(|value| value.to_str().ok()),
Some("no-store, no-cache, must-revalidate")
);
let content_type = response
.headers()
.get("content-type")
@@ -161,6 +168,13 @@ async fn gateway_serves_frontend_routes_and_assets_without_shadowing_public_api(
.await
.expect("spa request should succeed");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response
.headers()
.get("cache-control")
.and_then(|value| value.to_str().ok()),
Some("no-store, no-cache, must-revalidate")
);
let body = response.text().await.expect("spa body should be readable");
assert!(body.contains("Aether Frontend"));