624 lines
22 KiB
Python
624 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Optional, cast
|
|
|
|
from .errors import ApiError
|
|
from .models import *
|
|
|
|
|
|
class TransportSingBoxApiMixin:
|
|
def transport_singbox_profiles_get(
|
|
self,
|
|
*,
|
|
enabled_only: bool = False,
|
|
mode: str = "",
|
|
protocol: str = "",
|
|
) -> SingBoxProfilesState:
|
|
params: Dict[str, Any] = {}
|
|
if enabled_only:
|
|
params["enabled_only"] = "true"
|
|
mode_v = str(mode or "").strip().lower()
|
|
if mode_v:
|
|
params["mode"] = mode_v
|
|
protocol_v = str(protocol or "").strip().lower()
|
|
if protocol_v:
|
|
params["protocol"] = protocol_v
|
|
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"GET",
|
|
"/api/v1/transport/singbox/profiles",
|
|
params=(params or None),
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profiles_state(data)
|
|
|
|
def transport_singbox_profile_get(self, profile_id: str) -> SingBoxProfile:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(self._request("GET", f"/api/v1/transport/singbox/profiles/{pid}")) or {},
|
|
)
|
|
snap = self._parse_singbox_profiles_state(data)
|
|
if snap.item is None:
|
|
raise ApiError(
|
|
"API returned malformed singbox profile payload",
|
|
"GET",
|
|
self._url(f"/api/v1/transport/singbox/profiles/{pid}"),
|
|
)
|
|
return snap.item
|
|
|
|
def transport_singbox_profile_create(
|
|
self,
|
|
*,
|
|
profile_id: str = "",
|
|
name: str = "",
|
|
mode: str = "raw",
|
|
protocol: str = "",
|
|
enabled: Optional[bool] = None,
|
|
schema_version: int = 1,
|
|
typed: Optional[Dict[str, Any]] = None,
|
|
raw_config: Optional[Dict[str, Any]] = None,
|
|
meta: Optional[Dict[str, Any]] = None,
|
|
secrets: Optional[Dict[str, str]] = None,
|
|
) -> SingBoxProfilesState:
|
|
payload: Dict[str, Any] = {
|
|
"id": str(profile_id or "").strip(),
|
|
"name": str(name or "").strip(),
|
|
"mode": str(mode or "raw").strip().lower(),
|
|
"protocol": str(protocol or "").strip().lower(),
|
|
"schema_version": int(schema_version or 1),
|
|
}
|
|
if enabled is not None:
|
|
payload["enabled"] = bool(enabled)
|
|
if typed is not None:
|
|
payload["typed"] = cast(Dict[str, Any], typed)
|
|
if raw_config is not None:
|
|
payload["raw_config"] = cast(Dict[str, Any], raw_config)
|
|
if meta is not None:
|
|
payload["meta"] = cast(Dict[str, Any], meta)
|
|
if secrets is not None:
|
|
payload["secrets"] = {
|
|
str(k).strip(): str(v)
|
|
for k, v in cast(Dict[str, Any], secrets).items()
|
|
if str(k).strip()
|
|
}
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"POST",
|
|
"/api/v1/transport/singbox/profiles",
|
|
json_body=payload,
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profiles_state(data)
|
|
|
|
def transport_singbox_profile_patch(
|
|
self,
|
|
profile_id: str,
|
|
*,
|
|
base_revision: int = 0,
|
|
name: Optional[str] = None,
|
|
mode: Optional[str] = None,
|
|
protocol: Optional[str] = None,
|
|
enabled: Optional[bool] = None,
|
|
schema_version: Optional[int] = None,
|
|
typed: Optional[Dict[str, Any]] = None,
|
|
raw_config: Optional[Dict[str, Any]] = None,
|
|
meta: Optional[Dict[str, Any]] = None,
|
|
secrets: Optional[Dict[str, str]] = None,
|
|
clear_secrets: bool = False,
|
|
) -> SingBoxProfilesState:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
|
|
payload: Dict[str, Any] = {}
|
|
if int(base_revision or 0) > 0:
|
|
payload["base_revision"] = int(base_revision)
|
|
if name is not None:
|
|
payload["name"] = str(name)
|
|
if mode is not None:
|
|
payload["mode"] = str(mode or "").strip().lower()
|
|
if protocol is not None:
|
|
payload["protocol"] = str(protocol or "").strip().lower()
|
|
if enabled is not None:
|
|
payload["enabled"] = bool(enabled)
|
|
if schema_version is not None:
|
|
payload["schema_version"] = int(schema_version)
|
|
if typed is not None:
|
|
payload["typed"] = cast(Dict[str, Any], typed)
|
|
if raw_config is not None:
|
|
payload["raw_config"] = cast(Dict[str, Any], raw_config)
|
|
if meta is not None:
|
|
payload["meta"] = cast(Dict[str, Any], meta)
|
|
if secrets is not None:
|
|
payload["secrets"] = {
|
|
str(k).strip(): str(v)
|
|
for k, v in cast(Dict[str, Any], secrets).items()
|
|
if str(k).strip()
|
|
}
|
|
if clear_secrets:
|
|
payload["clear_secrets"] = True
|
|
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"PATCH",
|
|
f"/api/v1/transport/singbox/profiles/{pid}",
|
|
json_body=payload,
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profiles_state(data)
|
|
|
|
def transport_singbox_profile_render(
|
|
self,
|
|
profile_id: str,
|
|
*,
|
|
base_revision: int = 0,
|
|
check_binary: Optional[bool] = None,
|
|
persist: Optional[bool] = None,
|
|
) -> SingBoxProfileRenderResult:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
|
|
payload: Dict[str, Any] = {}
|
|
if int(base_revision or 0) > 0:
|
|
payload["base_revision"] = int(base_revision)
|
|
if check_binary is not None:
|
|
payload["check_binary"] = bool(check_binary)
|
|
if persist is not None:
|
|
payload["persist"] = bool(persist)
|
|
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"POST",
|
|
f"/api/v1/transport/singbox/profiles/{pid}/render",
|
|
json_body=payload,
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profile_render(data, fallback_id=pid)
|
|
|
|
def transport_singbox_profile_validate(
|
|
self,
|
|
profile_id: str,
|
|
*,
|
|
base_revision: int = 0,
|
|
check_binary: Optional[bool] = None,
|
|
) -> SingBoxProfileValidateResult:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
|
|
payload: Dict[str, Any] = {}
|
|
if int(base_revision or 0) > 0:
|
|
payload["base_revision"] = int(base_revision)
|
|
if check_binary is not None:
|
|
payload["check_binary"] = bool(check_binary)
|
|
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"POST",
|
|
f"/api/v1/transport/singbox/profiles/{pid}/validate",
|
|
json_body=payload,
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profile_validate(data, fallback_id=pid)
|
|
|
|
def transport_singbox_profile_apply(
|
|
self,
|
|
profile_id: str,
|
|
*,
|
|
client_id: str = "",
|
|
config_path: str = "",
|
|
restart: Optional[bool] = None,
|
|
skip_runtime: bool = False,
|
|
check_binary: Optional[bool] = None,
|
|
base_revision: int = 0,
|
|
) -> SingBoxProfileApplyResult:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
|
|
payload: Dict[str, Any] = {}
|
|
if int(base_revision or 0) > 0:
|
|
payload["base_revision"] = int(base_revision)
|
|
cid = str(client_id or "").strip()
|
|
if cid:
|
|
payload["client_id"] = cid
|
|
path = str(config_path or "").strip()
|
|
if path:
|
|
payload["config_path"] = path
|
|
if restart is not None:
|
|
payload["restart"] = bool(restart)
|
|
if bool(skip_runtime):
|
|
payload["skip_runtime"] = True
|
|
if check_binary is not None:
|
|
payload["check_binary"] = bool(check_binary)
|
|
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"POST",
|
|
f"/api/v1/transport/singbox/profiles/{pid}/apply",
|
|
json_body=payload,
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profile_apply(data, fallback_id=pid, fallback_client=cid)
|
|
|
|
def transport_singbox_profile_rollback(
|
|
self,
|
|
profile_id: str,
|
|
*,
|
|
base_revision: int = 0,
|
|
client_id: str = "",
|
|
config_path: str = "",
|
|
history_id: str = "",
|
|
restart: Optional[bool] = None,
|
|
skip_runtime: bool = False,
|
|
) -> SingBoxProfileRollbackResult:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
|
|
payload: Dict[str, Any] = {}
|
|
if int(base_revision or 0) > 0:
|
|
payload["base_revision"] = int(base_revision)
|
|
cid = str(client_id or "").strip()
|
|
if cid:
|
|
payload["client_id"] = cid
|
|
path = str(config_path or "").strip()
|
|
if path:
|
|
payload["config_path"] = path
|
|
hid = str(history_id or "").strip()
|
|
if hid:
|
|
payload["history_id"] = hid
|
|
if restart is not None:
|
|
payload["restart"] = bool(restart)
|
|
if bool(skip_runtime):
|
|
payload["skip_runtime"] = True
|
|
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"POST",
|
|
f"/api/v1/transport/singbox/profiles/{pid}/rollback",
|
|
json_body=payload,
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profile_rollback(data, fallback_id=pid, fallback_client=cid)
|
|
|
|
def transport_singbox_profile_history(
|
|
self,
|
|
profile_id: str,
|
|
*,
|
|
limit: int = 20,
|
|
) -> SingBoxProfileHistoryResult:
|
|
pid = str(profile_id or "").strip()
|
|
if not pid:
|
|
raise ValueError("missing singbox profile id")
|
|
lim = int(limit or 0)
|
|
if lim <= 0:
|
|
lim = 20
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"GET",
|
|
f"/api/v1/transport/singbox/profiles/{pid}/history",
|
|
params={"limit": str(lim)},
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_singbox_profile_history(data, fallback_id=pid)
|
|
|
|
# DNS / SmartDNS
|
|
|
|
def _parse_singbox_profile_issue(self, row: Any) -> Optional[SingBoxProfileIssue]:
|
|
if not isinstance(row, dict):
|
|
return None
|
|
return SingBoxProfileIssue(
|
|
field=str(row.get("field") or "").strip(),
|
|
severity=str(row.get("severity") or "").strip().lower(),
|
|
code=str(row.get("code") or "").strip(),
|
|
message=str(row.get("message") or "").strip(),
|
|
)
|
|
|
|
def _parse_singbox_profile_diff(self, raw: Any) -> SingBoxProfileRenderDiff:
|
|
data = raw if isinstance(raw, dict) else {}
|
|
return SingBoxProfileRenderDiff(
|
|
added=self._to_int(data.get("added")),
|
|
changed=self._to_int(data.get("changed")),
|
|
removed=self._to_int(data.get("removed")),
|
|
)
|
|
|
|
def _parse_singbox_profile(self, raw: Any) -> Optional[SingBoxProfile]:
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
pid = str(raw.get("id") or "").strip()
|
|
if not pid:
|
|
return None
|
|
|
|
typed_raw = raw.get("typed") or {}
|
|
if not isinstance(typed_raw, dict):
|
|
typed_raw = {}
|
|
raw_cfg = raw.get("raw_config") or {}
|
|
if not isinstance(raw_cfg, dict):
|
|
raw_cfg = {}
|
|
meta_raw = raw.get("meta") or {}
|
|
if not isinstance(meta_raw, dict):
|
|
meta_raw = {}
|
|
masked_raw = raw.get("secrets_masked") or {}
|
|
if not isinstance(masked_raw, dict):
|
|
masked_raw = {}
|
|
masked: Dict[str, str] = {}
|
|
for k, v in masked_raw.items():
|
|
key = str(k or "").strip()
|
|
if not key:
|
|
continue
|
|
masked[key] = str(v or "")
|
|
|
|
return SingBoxProfile(
|
|
id=pid,
|
|
name=str(raw.get("name") or "").strip(),
|
|
mode=str(raw.get("mode") or "").strip().lower(),
|
|
protocol=str(raw.get("protocol") or "").strip().lower(),
|
|
enabled=bool(raw.get("enabled", False)),
|
|
schema_version=self._to_int(raw.get("schema_version"), default=1),
|
|
profile_revision=self._to_int(raw.get("profile_revision")),
|
|
render_revision=self._to_int(raw.get("render_revision")),
|
|
last_validated_at=str(raw.get("last_validated_at") or "").strip(),
|
|
last_applied_at=str(raw.get("last_applied_at") or "").strip(),
|
|
last_error=str(raw.get("last_error") or "").strip(),
|
|
typed=cast(Dict[str, Any], typed_raw),
|
|
raw_config=cast(Dict[str, Any], raw_cfg),
|
|
meta=cast(Dict[str, Any], meta_raw),
|
|
has_secrets=bool(raw.get("has_secrets", False)),
|
|
secrets_masked=masked,
|
|
created_at=str(raw.get("created_at") or "").strip(),
|
|
updated_at=str(raw.get("updated_at") or "").strip(),
|
|
)
|
|
|
|
def _parse_singbox_profiles_state(self, data: Dict[str, Any]) -> SingBoxProfilesState:
|
|
raw_items = data.get("items") or []
|
|
if not isinstance(raw_items, list):
|
|
raw_items = []
|
|
items: List[SingBoxProfile] = []
|
|
for row in raw_items:
|
|
p = self._parse_singbox_profile(row)
|
|
if p is not None:
|
|
items.append(p)
|
|
|
|
item = self._parse_singbox_profile(data.get("item"))
|
|
|
|
return SingBoxProfilesState(
|
|
ok=bool(data.get("ok", False)),
|
|
message=str(data.get("message") or "").strip(),
|
|
code=str(data.get("code") or "").strip(),
|
|
count=self._to_int(data.get("count"), default=len(items)),
|
|
active_profile_id=str(data.get("active_profile_id") or "").strip(),
|
|
items=items,
|
|
item=item,
|
|
)
|
|
|
|
def _parse_singbox_profile_validate(
|
|
self,
|
|
data: Dict[str, Any],
|
|
*,
|
|
fallback_id: str = "",
|
|
) -> SingBoxProfileValidateResult:
|
|
raw_errors = data.get("errors") or []
|
|
if not isinstance(raw_errors, list):
|
|
raw_errors = []
|
|
raw_warnings = data.get("warnings") or []
|
|
if not isinstance(raw_warnings, list):
|
|
raw_warnings = []
|
|
|
|
errors: List[SingBoxProfileIssue] = []
|
|
for row in raw_errors:
|
|
issue = self._parse_singbox_profile_issue(row)
|
|
if issue is not None:
|
|
errors.append(issue)
|
|
|
|
warnings: List[SingBoxProfileIssue] = []
|
|
for row in raw_warnings:
|
|
issue = self._parse_singbox_profile_issue(row)
|
|
if issue is not None:
|
|
warnings.append(issue)
|
|
|
|
return SingBoxProfileValidateResult(
|
|
ok=bool(data.get("ok", False)),
|
|
message=str(data.get("message") or "").strip(),
|
|
code=str(data.get("code") or "").strip(),
|
|
profile_id=str(data.get("profile_id") or fallback_id).strip(),
|
|
profile_revision=self._to_int(data.get("profile_revision")),
|
|
valid=bool(data.get("valid", False)),
|
|
errors=errors,
|
|
warnings=warnings,
|
|
render_digest=str(data.get("render_digest") or "").strip(),
|
|
diff=self._parse_singbox_profile_diff(data.get("diff")),
|
|
)
|
|
|
|
def _parse_singbox_profile_apply(
|
|
self,
|
|
data: Dict[str, Any],
|
|
*,
|
|
fallback_id: str = "",
|
|
fallback_client: str = "",
|
|
) -> SingBoxProfileApplyResult:
|
|
raw_errors = data.get("errors") or []
|
|
if not isinstance(raw_errors, list):
|
|
raw_errors = []
|
|
raw_warnings = data.get("warnings") or []
|
|
if not isinstance(raw_warnings, list):
|
|
raw_warnings = []
|
|
|
|
errors: List[SingBoxProfileIssue] = []
|
|
for row in raw_errors:
|
|
issue = self._parse_singbox_profile_issue(row)
|
|
if issue is not None:
|
|
errors.append(issue)
|
|
|
|
warnings: List[SingBoxProfileIssue] = []
|
|
for row in raw_warnings:
|
|
issue = self._parse_singbox_profile_issue(row)
|
|
if issue is not None:
|
|
warnings.append(issue)
|
|
|
|
return SingBoxProfileApplyResult(
|
|
ok=bool(data.get("ok", False)),
|
|
message=str(data.get("message") or "").strip(),
|
|
code=str(data.get("code") or "").strip(),
|
|
profile_id=str(data.get("profile_id") or fallback_id).strip(),
|
|
client_id=str(data.get("client_id") or fallback_client).strip(),
|
|
config_path=str(data.get("config_path") or "").strip(),
|
|
profile_revision=self._to_int(data.get("profile_revision")),
|
|
render_revision=self._to_int(data.get("render_revision")),
|
|
last_applied_at=str(data.get("last_applied_at") or "").strip(),
|
|
render_path=str(data.get("render_path") or "").strip(),
|
|
render_digest=str(data.get("render_digest") or "").strip(),
|
|
rollback_available=bool(data.get("rollback_available", False)),
|
|
valid=bool(data.get("valid", False)),
|
|
errors=errors,
|
|
warnings=warnings,
|
|
diff=self._parse_singbox_profile_diff(data.get("diff")),
|
|
)
|
|
|
|
def _parse_singbox_profile_render(
|
|
self,
|
|
data: Dict[str, Any],
|
|
*,
|
|
fallback_id: str = "",
|
|
) -> SingBoxProfileRenderResult:
|
|
raw_errors = data.get("errors") or []
|
|
if not isinstance(raw_errors, list):
|
|
raw_errors = []
|
|
raw_warnings = data.get("warnings") or []
|
|
if not isinstance(raw_warnings, list):
|
|
raw_warnings = []
|
|
raw_config = data.get("config") or {}
|
|
if not isinstance(raw_config, dict):
|
|
raw_config = {}
|
|
|
|
errors: List[SingBoxProfileIssue] = []
|
|
for row in raw_errors:
|
|
issue = self._parse_singbox_profile_issue(row)
|
|
if issue is not None:
|
|
errors.append(issue)
|
|
|
|
warnings: List[SingBoxProfileIssue] = []
|
|
for row in raw_warnings:
|
|
issue = self._parse_singbox_profile_issue(row)
|
|
if issue is not None:
|
|
warnings.append(issue)
|
|
|
|
return SingBoxProfileRenderResult(
|
|
ok=bool(data.get("ok", False)),
|
|
message=str(data.get("message") or "").strip(),
|
|
code=str(data.get("code") or "").strip(),
|
|
profile_id=str(data.get("profile_id") or fallback_id).strip(),
|
|
profile_revision=self._to_int(data.get("profile_revision")),
|
|
render_revision=self._to_int(data.get("render_revision")),
|
|
render_path=str(data.get("render_path") or "").strip(),
|
|
render_digest=str(data.get("render_digest") or "").strip(),
|
|
changed=bool(data.get("changed", False)),
|
|
valid=bool(data.get("valid", False)),
|
|
errors=errors,
|
|
warnings=warnings,
|
|
diff=self._parse_singbox_profile_diff(data.get("diff")),
|
|
config=cast(Dict[str, Any], raw_config),
|
|
)
|
|
|
|
def _parse_singbox_profile_rollback(
|
|
self,
|
|
data: Dict[str, Any],
|
|
*,
|
|
fallback_id: str = "",
|
|
fallback_client: str = "",
|
|
) -> SingBoxProfileRollbackResult:
|
|
return SingBoxProfileRollbackResult(
|
|
ok=bool(data.get("ok", False)),
|
|
message=str(data.get("message") or "").strip(),
|
|
code=str(data.get("code") or "").strip(),
|
|
profile_id=str(data.get("profile_id") or fallback_id).strip(),
|
|
client_id=str(data.get("client_id") or fallback_client).strip(),
|
|
config_path=str(data.get("config_path") or "").strip(),
|
|
history_id=str(data.get("history_id") or "").strip(),
|
|
profile_revision=self._to_int(data.get("profile_revision")),
|
|
last_applied_at=str(data.get("last_applied_at") or "").strip(),
|
|
)
|
|
|
|
def _parse_singbox_profile_history_entry(self, raw: Any) -> Optional[SingBoxProfileHistoryEntry]:
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
hid = str(raw.get("id") or "").strip()
|
|
if not hid:
|
|
return None
|
|
return SingBoxProfileHistoryEntry(
|
|
id=hid,
|
|
at=str(raw.get("at") or "").strip(),
|
|
profile_id=str(raw.get("profile_id") or "").strip(),
|
|
action=str(raw.get("action") or "").strip().lower(),
|
|
status=str(raw.get("status") or "").strip().lower(),
|
|
code=str(raw.get("code") or "").strip(),
|
|
message=str(raw.get("message") or "").strip(),
|
|
profile_revision=self._to_int(raw.get("profile_revision")),
|
|
render_revision=self._to_int(raw.get("render_revision")),
|
|
render_digest=str(raw.get("render_digest") or "").strip(),
|
|
render_path=str(raw.get("render_path") or "").strip(),
|
|
client_id=str(raw.get("client_id") or "").strip(),
|
|
)
|
|
|
|
def _parse_singbox_profile_history(
|
|
self,
|
|
data: Dict[str, Any],
|
|
*,
|
|
fallback_id: str = "",
|
|
) -> SingBoxProfileHistoryResult:
|
|
raw_items = data.get("items") or []
|
|
if not isinstance(raw_items, list):
|
|
raw_items = []
|
|
items: List[SingBoxProfileHistoryEntry] = []
|
|
for row in raw_items:
|
|
entry = self._parse_singbox_profile_history_entry(row)
|
|
if entry is not None:
|
|
items.append(entry)
|
|
|
|
return SingBoxProfileHistoryResult(
|
|
ok=bool(data.get("ok", False)),
|
|
message=str(data.get("message") or "").strip(),
|
|
code=str(data.get("code") or "").strip(),
|
|
profile_id=str(data.get("profile_id") or fallback_id).strip(),
|
|
count=self._to_int(data.get("count"), default=len(items)),
|
|
items=items,
|
|
)
|
|
|