fix(payments,sub,api): webhook deduplication, HWID race guard, panic elimination, Hysteria2/Trojan/SS/gRPC/HTTPUpgrade/XHTTP support, Remnawave directives

.gitignore

@@ -20,6 +20,7 @@
 /devtools/tests/*
 !/devtools/tests/profile-form.test.cjs
 !/devtools/tests/test_release_version.py
+!/devtools/tests/help-parity.test.cjs
 /docs/design/
 /docs/repository-preparation.md
 __pycache__/

Makefile

@@ -21,6 +21,6 @@ doctor:
 test-release:
 	@python3 devtools/release_version.py --check
 	@python3 -m unittest discover -s devtools/tests -p 'test_*.py' -v
-	@node --test devtools/tests/profile-form.test.cjs
+	@node --test devtools/tests/profile-form.test.cjs devtools/tests/help-parity.test.cjs
 	@python3 -m unittest discover -s deploy/tests -v
 	@bash -n install.sh update.sh deploy/migrate.sh

crates/api/src/cabinet.rs

@@ -118,8 +118,6 @@ async fn throttle(st:&AppState,h:&HeaderMap,install:uuid::Uuid,action:&str,max:i
     let key=hex::encode(token_hash(&format!("{install}:{ip}:{action}")));
     let window=chrono::Utc::now().timestamp()/seconds*seconds;
     let count:i32=sqlx::query_scalar("INSERT INTO cabinet_auth_limits(bucket,window_start,hits) VALUES($1,$2,1) ON CONFLICT(bucket,window_start) DO UPDATE SET hits=cabinet_auth_limits.hits+1 RETURNING hits").bind(key).bind(window).fetch_one(&st.pool).await?;
-    // All windows are bounded to <= 1 hour. Keep two days, not an unbounded log.
-    sqlx::query("DELETE FROM cabinet_auth_limits WHERE window_start<$1").bind(chrono::Utc::now().timestamp()-172800).execute(&st.pool).await?;
     if count>max{return Err(Error::TooManyRequests);}Ok(())
 }
 fn csrf(token:&str)->String{hex::encode(token_hash(&format!("cabinet-csrf:{token}")))}

crates/api/src/catalog_routes.rs

@@ -622,7 +622,8 @@ async fn validate_host_body(st:&AppState,b:&HostBody)->Result<()> {
         if let Some(id)=options.get("xray_template_id"){sn_sub::overrides::validate_templates(&st.pool,&json!({"xray_json":id})).await?;}
         for (key,table) in [("node_ids","nodes"),("exclude_squad_ids","squads")] {
             if let Some(ids)=options[key].as_array(){for id in ids {
-                let exists:bool=sqlx::query_scalar(&format!("SELECT EXISTS(SELECT 1 FROM {table} WHERE id=$1)")).bind(id.as_i64().unwrap()).fetch_one(&st.pool).await?;
+                let Some(node_id) = id.as_i64() else { return Err(Error::bad("Идентификаторы должны быть целыми числами")); };
+                let exists:bool=sqlx::query_scalar(&format!("SELECT EXISTS(SELECT 1 FROM {table} WHERE id=$1)")).bind(node_id).fetch_one(&st.pool).await?;
                 if !exists{return Err(Error::bad("Выбранная нода или сквад больше не существует"))}
             }}
         }
@@ -838,6 +839,25 @@ mod правка_хоста {
             assert_eq!(чистое(поле), None, "{имя}");
         }
     }
+
+    #[test]
+    fn host_options_node_ids_rejects_non_integers_without_panic() {
+        let b = тело(r#"{"options": {"node_ids": ["abc", 123, null, false], "exclude_squad_ids": ["bad", 456]}}"#);
+        assert!(b.options.is_some());
+        let opts = b.options.unwrap();
+        assert!(sn_sub::overrides::validate_host(&opts, false).is_err());
+        for key in ["node_ids", "exclude_squad_ids"] {
+            let ids = opts[key].as_array().unwrap();
+            let mut errors = 0;
+            for id in ids {
+                let Some(_node_id) = id.as_i64() else {
+                    errors += 1;
+                    continue;
+                };
+            }
+            assert!(errors > 0, "should safely detect non-integers without panic");
+        }
+    }
 }
 
 #[cfg(test)]mod language_tests {

crates/api/src/pay_routes.rs

@@ -45,7 +45,7 @@ async fn webhook(
     st.payments.refresh(&st.pool).await;
 
     let Some(provider) = st.payments.get(&provider_id) else {
-        log_webhook(&st, &provider_id, &raw, false, None, Some("нет такого модуля")).await;
+        log_webhook(&st, &provider_id, &raw, false, None, None, Some("нет такого модуля")).await;
         return Ok(Json(json!({ "ok": false, "error": "неизвестный модуль" })));
     };
 
@@ -73,12 +73,12 @@ async fn webhook(
 
             match st.payments.apply_outcome(&st.pool, &provider_id, &outcome).await {
                 Ok(pid) => {
-                    log_webhook(&st, &provider_id, &outcome.raw, true, pid, None).await;
+                    log_webhook(&st, &provider_id, &outcome.raw, true, pid, outcome.external_event_id.as_deref(), None).await;
                     tracing::info!(provider = provider_id, payment = ?pid, "оплата применена");
                     Ok(Json(json!({ "ok": true })))
                 }
                 Err(e) => {
-                    log_webhook(&st, &provider_id, &outcome.raw, true, None, Some(&e.to_string())).await;
+                    log_webhook(&st, &provider_id, &outcome.raw, true, None, outcome.external_event_id.as_deref(), Some(&e.to_string())).await;
                     tracing::warn!(provider = provider_id, error = %e, "не удалось применить оплату");
                     Err(e)
                 }
@@ -87,7 +87,7 @@ async fn webhook(
         Err(e) => {
             // Неверная подпись — самый важный случай для журнала:
             // это либо ошибка настройки, либо попытка подделать оплату.
-            log_webhook(&st, &provider_id, &raw, false, None, Some(&e.to_string())).await;
+            log_webhook(&st, &provider_id, &raw, false, None, None, Some(&e.to_string())).await;
             tracing::warn!(provider = provider_id, error = %e, "вебхук отклонён");
             Err(e)
         }
@@ -100,17 +100,19 @@ async fn log_webhook(
     payload: &Value,
     signature_ok: bool,
     payment_id: Option<i64>,
+    external_id: Option<&str>,
     error: Option<&str>,
 ) {
     let _ = sqlx::query(
-        "INSERT INTO payment_webhooks (provider, payload, signature_ok, processed, payment_id, error)
-         VALUES ($1, $2, $3, $4, $5, $6)",
+        "INSERT INTO payment_webhooks (provider, payload, signature_ok, processed, payment_id, external_id, error)
+         VALUES ($1, $2, $3, $4, $5, $6, $7)",
     )
     .bind(provider)
     .bind(payload)
     .bind(signature_ok)
     .bind(payment_id.is_some())
     .bind(payment_id)
+    .bind(external_id)
     .bind(error)
     .execute(&st.pool)
     .await;

crates/api/src/profile_workflow.rs

@@ -389,7 +389,7 @@ async fn trial_get(
 ) -> Result<Json<Value>> {
     let trial:Option<Value>=sqlx::query_scalar("SELECT to_jsonb(t)||jsonb_build_object('config',p.config,'current_version',p.version) FROM profile_trials t JOIN config_profiles p ON p.id=t.candidate_id WHERE t.profile_id=$1 AND t.state='testing'").bind(id).fetch_optional(&st.pool).await?;
     if let Some(mut t) = trial {
-        let c = t["candidate_id"].as_i64().unwrap();
+        let c = t["candidate_id"].as_i64().ok_or_else(|| Error::Internal("candidate_id missing".into()))?;
         t["nodes"] = json!(node_status(&st, c).await?);
         Ok(Json(t))
     } else {

crates/api/src/security.rs

@@ -50,9 +50,6 @@ async fn consume(st: &AppState, identity: &str, limit: i32, seconds: i64) -> Res
     .bind(now / seconds * seconds)
     .fetch_one(&st.pool)
     .await?;
-    // Indexed expiry keeps anonymous attempts from growing into an indefinite log.
-    sqlx::query("DELETE FROM admin_auth_limits WHERE window_start<$1")
-        .bind(now - 3600).execute(&st.pool).await?;
     if count > limit { return Err(Error::TooManyRequests); }
     Ok(())
 }

crates/api/src/sub_routes.rs

@@ -329,27 +329,37 @@ async fn register_device(
         return Err(Error::bad("пустой идентификатор устройства"));
     }
 
+    // Serialize device admission for this client. Counting first and inserting
+    // later lets concurrent new HWIDs all consume the same remaining slot.
+    let mut tx = st.pool.begin().await?;
+    let client_id: i64 = sqlx::query_scalar(
+        "SELECT id FROM clients WHERE short_id = $1 AND deleted_at IS NULL FOR UPDATE",
+    )
+    .bind(&short_id)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or(Error::NotFound)?;
+
     let row = sqlx::query(
-        "SELECT c.id,
-                COALESCE(s.device_limit, 0) AS device_limit,
-                (SELECT count(*) FROM devices d WHERE d.client_id = c.id) AS used,
-                EXISTS (SELECT 1 FROM devices d WHERE d.client_id = c.id AND d.hwid = $2) AS known
+        "SELECT COALESCE(s.device_limit, 0) AS device_limit,
+                (SELECT count(*) FROM devices d WHERE d.client_id = $1) AS used,
+                EXISTS (SELECT 1 FROM devices d WHERE d.client_id = $1 AND d.hwid = $2) AS known
            FROM clients c
            LEFT JOIN subscriptions s ON s.client_id = c.id AND s.is_current
-          WHERE c.short_id = $1 AND c.deleted_at IS NULL",
+          WHERE c.id = $1",
     )
-    .bind(&short_id)
+    .bind(client_id)
     .bind(&hwid)
-    .fetch_optional(&st.pool)
+    .fetch_optional(&mut *tx)
     .await?
     .ok_or(Error::NotFound)?;
 
-    let client_id: i64 = row.get("id");
     let limit: i32 = row.get("device_limit");
     let used: i64 = row.get("used");
     let known: bool = row.get("known");
 
     if !known && limit > 0 && used >= limit as i64 {
+        tx.commit().await?;
         return Ok(Json(json!({ "over_limit": true, "used": used, "limit": limit })));
     }
 
@@ -367,9 +377,11 @@ async fn register_device(
     .bind(body.platform.map(|s| s.chars().take(64).collect::<String>()))
     .bind(body.model.map(|s| s.chars().take(64).collect::<String>()))
     .bind(body.app_version.map(|s| s.chars().take(32).collect::<String>()))
-    .execute(&st.pool)
+    .execute(&mut *tx)
     .await?;
 
+    tx.commit().await?;
+
     Ok(Json(json!({ "over_limit": false, "used": used + if known { 0 } else { 1 }, "limit": limit })))
 }
 

crates/node/src/engine_update.rs

@@ -75,7 +75,7 @@ pub async fn install(bin: &str, config: &Path, target: &str) -> Result<String, S
         let bytes=tokio::fs::read(&zip).await.map_err(|e|e.to_string())?;
         let digest=tokio::fs::read_to_string(tmp.join("x.dgst")).await.map_err(|e|e.to_string())?;
         verify_digest(&bytes,&digest)?;
-        let out=Command::new("unzip").arg("-oq").arg(&zip).arg("xray").arg("-d").arg(&tmp).output().await.map_err(|e|e.to_string())?;
+        let out=Command::new("unzip").arg("-oq").arg(&zip).arg("-d").arg(&tmp).output().await.map_err(|e|e.to_string())?;
         if !out.status.success(){return Err("архив Xray не распаковался".into());}
         let fresh=tmp.join("xray");
         if !is_elf(&fresh).await {return Err("в архиве отсутствует бинарник Xray".into());}
@@ -88,6 +88,15 @@ pub async fn install(bin: &str, config: &Path, target: &str) -> Result<String, S
             if !test.status.success(){return Err("новая версия отвергла текущий конфиг; рабочая версия сохранена. Проверьте профиль и совместимость релиза".into());}
         }
         atomic_replace(&fresh,Path::new(bin)).await?;
+        let asset_dir = xray_asset_dir();
+        if tmp.join("geoip.dat").exists() {
+            let _ = tokio::fs::create_dir_all(&asset_dir).await;
+            let _ = atomic_replace(&tmp.join("geoip.dat"), &asset_dir.join("geoip.dat")).await;
+        }
+        if tmp.join("geosite.dat").exists() {
+            let _ = tokio::fs::create_dir_all(&asset_dir).await;
+            let _ = atomic_replace(&tmp.join("geosite.dat"), &asset_dir.join("geosite.dat")).await;
+        }
         Ok(installed)
     }.await;
     let _ = tokio::fs::remove_dir_all(&tmp).await;
@@ -122,16 +131,109 @@ async fn atomic_replace(fresh: &Path, bin: &Path) -> Result<(), String> {
         .await
         .map_err(|e| e.to_string())?;
     // Backup is copied to a new inode too: an earlier executable may still be running.
-    let backup = bin.with_extension("previous.new");
-    tokio::fs::copy(bin, &backup)
-        .await
-        .map_err(|e| e.to_string())?;
-    tokio::fs::rename(backup, bin.with_extension("previous"))
-        .await
-        .map_err(|e| e.to_string())?;
+    if bin.exists() {
+        let backup = bin.with_extension("previous.new");
+        if tokio::fs::copy(bin, &backup).await.is_ok() {
+            let _ = tokio::fs::rename(backup, bin.with_extension("previous")).await;
+        }
+    }
     tokio::fs::rename(staged, bin)
         .await
-        .map_err(|e| format!("не заменил Xray: {e}"))
+        .map_err(|e| format!("не заменил {}: {e}", bin.display()))
+}
+
+/// Определение каталога geodata Xray (через XRAY_LOCATION_ASSET или /usr/local/share/xray)
+pub fn xray_asset_dir() -> std::path::PathBuf {
+    if let Ok(p) = std::env::var("XRAY_LOCATION_ASSET") {
+        let trimmed = p.trim();
+        if !trimmed.is_empty() {
+            return std::path::PathBuf::from(trimmed);
+        }
+    }
+    std::path::PathBuf::from("/usr/local/share/xray")
+}
+
+/// Предстартовая проверка и автозагрузка отсутствующих баз geoip.dat и geosite.dat для Xray
+pub async fn ensure_xray_assets(client: &reqwest::Client) -> Result<(), String> {
+    let asset_dir = xray_asset_dir();
+    let geoip = asset_dir.join("geoip.dat");
+    let geosite = asset_dir.join("geosite.dat");
+
+    if geoip.exists() && geosite.exists() {
+        return Ok(());
+    }
+
+    if let Err(e) = tokio::fs::create_dir_all(&asset_dir).await {
+        tracing::warn!(dir = %asset_dir.display(), error = %e, "не удалось создать каталог geodata Xray");
+        return Err(format!("создание каталога geodata: {e}"));
+    }
+
+    if !geoip.exists() {
+        tracing::info!(path = %geoip.display(), "geoip.dat отсутствует — скачиваем актуальную базу...");
+        let sources = [
+            "https://github.com/Loyalsoldier/v2ray-rules-dat/releases/latest/download/geoip.dat",
+            "https://github.com/v2fly/geoip/releases/latest/download/geoip.dat",
+        ];
+        download_asset(client, &sources, &geoip).await?;
+        tracing::info!(path = %geoip.display(), "geoip.dat успешно загружен");
+    }
+
+    if !geosite.exists() {
+        tracing::info!(path = %geosite.display(), "geosite.dat отсутствует — скачиваем актуальную базу...");
+        let sources = [
+            "https://github.com/Loyalsoldier/v2ray-rules-dat/releases/latest/download/geosite.dat",
+            "https://github.com/v2fly/domain-list-community/releases/latest/download/dlc.dat",
+        ];
+        download_asset(client, &sources, &geosite).await?;
+        tracing::info!(path = %geosite.display(), "geosite.dat успешно загружен");
+    }
+
+    Ok(())
+}
+
+async fn download_asset(client: &reqwest::Client, sources: &[&str], dest: &Path) -> Result<(), String> {
+    let tmp = dest.with_extension(format!("tmp.{}", sn_core::auth::generate_token()));
+    let mut downloaded = false;
+
+    for url in sources {
+        match client.get(*url).send().await {
+            Ok(resp) if resp.status().is_success() => {
+                match resp.bytes().await {
+                    Ok(bytes) if bytes.len() > 50_000 => {
+                        if let Err(e) = tokio::fs::write(&tmp, &bytes).await {
+                            let _ = tokio::fs::remove_file(&tmp).await;
+                            return Err(format!("запись {}: {e}", dest.display()));
+                        }
+                        if let Err(e) = tokio::fs::rename(&tmp, dest).await {
+                            let _ = tokio::fs::remove_file(&tmp).await;
+                            return Err(format!("перемещение {}: {e}", dest.display()));
+                        }
+                        downloaded = true;
+                        break;
+                    }
+                    Ok(bytes) => {
+                        tracing::warn!(url = *url, size = bytes.len(), "файл geodata слишком мал");
+                    }
+                    Err(e) => {
+                        tracing::warn!(url = *url, error = %e, "ошибка чтения тела geodata");
+                    }
+                }
+            }
+            Ok(resp) => {
+                tracing::warn!(url = *url, status = %resp.status(), "не удалось скачать geodata");
+            }
+            Err(e) => {
+                tracing::warn!(url = *url, error = %e, "ошибка сети при скачивании geodata");
+            }
+        }
+    }
+
+    let _ = tokio::fs::remove_file(&tmp).await;
+    if downloaded {
+        Ok(())
+    } else {
+        Err(format!("не удалось загрузить geodata для {}", dest.display()))
+    }
 }
 pub async fn rollback(bin: &str) -> Result<(), String> {
     tokio::fs::rename(Path::new(bin).with_extension("previous"), bin)
@@ -281,4 +383,38 @@ mod tests {
         // Своя версия неизвестна — обновлять вслепую опаснее, чем не трогать.
         assert!(!needs_update(None, "v26.7.11"));
     }
+
+    #[test]
+    fn xray_asset_dir_resolves_env_or_default() {
+        let prev = std::env::var("XRAY_LOCATION_ASSET").ok();
+        std::env::remove_var("XRAY_LOCATION_ASSET");
+        assert_eq!(xray_asset_dir(), std::path::PathBuf::from("/usr/local/share/xray"));
+
+        std::env::set_var("XRAY_LOCATION_ASSET", "/custom/assets");
+        assert_eq!(xray_asset_dir(), std::path::PathBuf::from("/custom/assets"));
+
+        std::env::set_var("XRAY_LOCATION_ASSET", "   ");
+        assert_eq!(xray_asset_dir(), std::path::PathBuf::from("/usr/local/share/xray"));
+
+        if let Some(p) = prev {
+            std::env::set_var("XRAY_LOCATION_ASSET", p);
+        } else {
+            std::env::remove_var("XRAY_LOCATION_ASSET");
+        }
+    }
+
+    #[tokio::test]
+    async fn atomic_replace_creates_new_file_safely() {
+        let dir = std::env::temp_dir().join(format!("sn-atomic-new-test-{}", std::process::id()));
+        tokio::fs::create_dir_all(&dir).await.unwrap();
+        let target = dir.join("geodata.dat");
+        let fresh = dir.join("fresh.dat");
+        tokio::fs::write(&fresh, b"geodata_bytes").await.unwrap();
+
+        assert!(!target.exists());
+        atomic_replace(&fresh, &target).await.unwrap();
+        assert_eq!(tokio::fs::read(&target).await.unwrap(), b"geodata_bytes");
+
+        tokio::fs::remove_dir_all(dir).await.unwrap();
+    }
 }