feat: install instruction inside web app
This commit is contained in:
3252a8
@@ -1,5 +1,11 @@
|
||||
# ruff: noqa: F401,F403,F405,I001
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
from .webapp_runtime import refresh_webapp_runtime_after_settings_change
|
||||
|
||||
from config.subscription_guides_config import (
|
||||
SubscriptionGuidesConfigError,
|
||||
subscription_guides_admin_config_json,
|
||||
)
|
||||
|
||||
|
||||
async def admin_settings_get_route(request: web.Request) -> web.Response:
|
||||
@@ -26,12 +32,25 @@ async def admin_settings_get_route(request: web.Request) -> web.Response:
|
||||
override = overrides_by_key.get(key)
|
||||
value = current_value(settings, key)
|
||||
is_secret = bool(field.get("secret"))
|
||||
overridden = bool(override)
|
||||
source = None
|
||||
read_error = None
|
||||
if key == "SUBSCRIPTION_PAGE_CONFIG_JSON":
|
||||
try:
|
||||
value, source = subscription_guides_admin_config_json(settings)
|
||||
overridden = source == "admin_json"
|
||||
except SubscriptionGuidesConfigError as exc:
|
||||
read_error = str(exc)
|
||||
response_field = {
|
||||
**field,
|
||||
"value": "" if is_secret else value,
|
||||
"overridden": bool(override),
|
||||
"overridden": overridden,
|
||||
"updated_at": override.get("updated_at") if override else None,
|
||||
}
|
||||
if source:
|
||||
response_field["source"] = source
|
||||
if read_error:
|
||||
response_field["read_error"] = read_error
|
||||
if is_secret:
|
||||
response_field["has_value"] = bool(value)
|
||||
sections[section_id]["fields"].append(response_field)
|
||||
@@ -51,6 +70,12 @@ async def admin_settings_patch_route(request: web.Request) -> web.Response:
|
||||
return _error(400, "invalid_updates")
|
||||
if not isinstance(deletes, list):
|
||||
return _error(400, "invalid_deletes")
|
||||
if "SUBSCRIPTION_PAGE_CONFIG_JSON" in updates and not str(
|
||||
updates.get("SUBSCRIPTION_PAGE_CONFIG_JSON") or ""
|
||||
).strip():
|
||||
updates = dict(updates)
|
||||
updates.pop("SUBSCRIPTION_PAGE_CONFIG_JSON", None)
|
||||
deletes = [*deletes, "SUBSCRIPTION_PAGE_CONFIG_JSON"]
|
||||
|
||||
result = await update_overrides(
|
||||
settings,
|
||||
@@ -65,26 +90,6 @@ async def admin_settings_patch_route(request: web.Request) -> web.Response:
|
||||
status=400,
|
||||
)
|
||||
|
||||
# Bust the public webapp settings cache so users see new values immediately.
|
||||
cache = request.app.get("webapp_settings_cache")
|
||||
if isinstance(cache, dict):
|
||||
cache["ts"] = 0.0
|
||||
cache["data"] = {}
|
||||
if (
|
||||
"WEBAPP_LOGO_URL" in updates
|
||||
or "WEBAPP_LOGO_URL" in deletes
|
||||
or "WEBAPP_LOGO_USE_EMOJI" in updates
|
||||
or "WEBAPP_LOGO_USE_EMOJI" in deletes
|
||||
or "WEBAPP_FAVICON_URL" in updates
|
||||
or "WEBAPP_FAVICON_URL" in deletes
|
||||
or "WEBAPP_FAVICON_USE_CUSTOM" in updates
|
||||
or "WEBAPP_FAVICON_USE_CUSTOM" in deletes
|
||||
or "WEBAPP_LOGO_FAVICON_URL" in updates
|
||||
or "WEBAPP_LOGO_FAVICON_URL" in deletes
|
||||
):
|
||||
request.app["webapp_logo_cache"] = None
|
||||
from bot.app.web.admin_api_impl.themes import prune_unused_appearance_assets
|
||||
|
||||
prune_unused_appearance_assets(settings)
|
||||
await refresh_webapp_runtime_after_settings_change(request, updates=updates, deletes=deletes)
|
||||
|
||||
return _ok({"applied": result.get("applied", 0), "reverted": result.get("reverted", 0)})
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
# ruff: noqa: F401,F403,F405,I001
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
from .webapp_runtime import refresh_webapp_runtime_after_settings_change
|
||||
|
||||
|
||||
async def admin_tariffs_get_route(request: web.Request) -> web.Response:
|
||||
@@ -55,9 +56,6 @@ async def admin_tariffs_save_route(request: web.Request) -> web.Response:
|
||||
logger.exception("Failed to write tariffs config to %s", path)
|
||||
return _error(500, "write_failed", str(exc))
|
||||
|
||||
cache = request.app.get("webapp_settings_cache")
|
||||
if isinstance(cache, dict):
|
||||
cache["ts"] = 0.0
|
||||
cache["data"] = {}
|
||||
await refresh_webapp_runtime_after_settings_change(request, updates={}, deletes=[])
|
||||
|
||||
return _ok({"exists": True, "path": str(path), "catalog": _tariffs_config_payload(config)})
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
# ruff: noqa: F401,F403,F405,I001
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
from .webapp_runtime import refresh_webapp_runtime_after_settings_change
|
||||
|
||||
import asyncio
|
||||
import hashlib
|
||||
@@ -213,12 +214,7 @@ async def _persist_appearance_upload(
|
||||
logger.warning("Failed to persist uploaded appearance asset settings: %s", result)
|
||||
return False
|
||||
|
||||
cache = request.app.get("webapp_settings_cache")
|
||||
if isinstance(cache, dict):
|
||||
cache["ts"] = 0.0
|
||||
cache["data"] = {}
|
||||
request.app["webapp_logo_cache"] = None
|
||||
prune_unused_appearance_assets(settings)
|
||||
await refresh_webapp_runtime_after_settings_change(request, updates=updates, deletes=[])
|
||||
return True
|
||||
|
||||
|
||||
@@ -484,10 +480,7 @@ async def admin_themes_save_route(request: web.Request) -> web.Response:
|
||||
logger.exception("Failed to write webapp themes to %s", settings.WEBAPP_THEMES_DIR)
|
||||
return _error(500, "write_failed", str(exc))
|
||||
|
||||
cache = request.app.get("webapp_settings_cache")
|
||||
if isinstance(cache, dict):
|
||||
cache["ts"] = 0.0
|
||||
cache["data"] = {}
|
||||
await refresh_webapp_runtime_after_settings_change(request, updates={}, deletes=[])
|
||||
|
||||
return _ok(
|
||||
{
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Any
|
||||
|
||||
from bot.app.web.webapp.cache_helpers import (
|
||||
invalidate_all_webapp_user_payloads,
|
||||
reset_subscription_guides_cache,
|
||||
reset_webapp_settings_cache,
|
||||
)
|
||||
|
||||
WEBAPP_APPEARANCE_SETTING_KEYS = frozenset(
|
||||
{
|
||||
"WEBAPP_LOGO_URL",
|
||||
"WEBAPP_LOGO_USE_EMOJI",
|
||||
"WEBAPP_LOGO_EMOJI",
|
||||
"WEBAPP_LOGO_EMOJI_FONT",
|
||||
"WEBAPP_FAVICON_URL",
|
||||
"WEBAPP_FAVICON_USE_CUSTOM",
|
||||
"WEBAPP_LOGO_FAVICON_URL",
|
||||
}
|
||||
)
|
||||
|
||||
WEBAPP_DEVICE_PAYLOAD_SETTING_KEYS = frozenset(
|
||||
{
|
||||
"MY_DEVICES_SECTION_ENABLED",
|
||||
"USER_HWID_DEVICE_LIMIT",
|
||||
"USER_TRAFFIC_LIMIT_GB",
|
||||
"USER_TRAFFIC_STRATEGY",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def changed_setting_keys(
|
||||
updates: Mapping[str, Any] | None = None,
|
||||
deletes: Sequence[Any] | None = None,
|
||||
) -> set[str]:
|
||||
keys = {str(key) for key in (updates or {}).keys()}
|
||||
keys.update(str(key) for key in (deletes or []) if key is not None)
|
||||
return keys
|
||||
|
||||
|
||||
async def refresh_webapp_runtime_after_settings_change(
|
||||
request: Any,
|
||||
*,
|
||||
updates: Mapping[str, Any] | None = None,
|
||||
deletes: Sequence[Any] | None = None,
|
||||
include_user_payloads: bool = True,
|
||||
) -> None:
|
||||
settings = request.app["settings"]
|
||||
keys = changed_setting_keys(updates, deletes)
|
||||
|
||||
reset_webapp_settings_cache(request.app)
|
||||
reset_subscription_guides_cache(request.app)
|
||||
|
||||
if include_user_payloads:
|
||||
await invalidate_all_webapp_user_payloads(
|
||||
settings,
|
||||
include_devices=bool(keys & WEBAPP_DEVICE_PAYLOAD_SETTING_KEYS),
|
||||
)
|
||||
|
||||
if keys & WEBAPP_APPEARANCE_SETTING_KEYS:
|
||||
request.app["webapp_logo_cache"] = None
|
||||
from bot.app.web.admin_api_impl.themes import prune_unused_appearance_assets
|
||||
|
||||
prune_unused_appearance_assets(settings)
|
||||
@@ -16,7 +16,7 @@ from typing import Any, List, Optional, Tuple
|
||||
@dataclass(frozen=True)
|
||||
class SettingField:
|
||||
key: str
|
||||
type: str # "string" | "int" | "float" | "bool" | "text" | "url" | "color" | "icon"
|
||||
type: str # "string" | "int" | "float" | "bool" | "text" | "url" | "color" | "icon" | "json"
|
||||
section: str
|
||||
label: str
|
||||
description: str = ""
|
||||
@@ -160,6 +160,49 @@ SETTINGS_MANIFEST: List[SettingField] = [
|
||||
SettingField("WEBAPP_FAVICON_URL", "url", "appearance", "URL отдельной favicon"),
|
||||
SettingField("WEBAPP_LOGO_FAVICON_URL", "url", "appearance", "Favicon из логотипа"),
|
||||
SettingField("WEBAPP_ENABLED", "bool", "appearance", "Web App включён"),
|
||||
SettingField(
|
||||
"SUBSCRIPTION_GUIDES_ENABLED",
|
||||
"bool",
|
||||
"subscription_guides",
|
||||
"Embedded install guides",
|
||||
"Open install instructions inside the Web App instead of an external connect page.",
|
||||
),
|
||||
SettingField(
|
||||
"SUBSCRIPTION_PAGE_CONFIG_PANEL_ENABLED",
|
||||
"bool",
|
||||
"subscription_guides",
|
||||
"Use Remnawave Panel config",
|
||||
(
|
||||
"Fetch Subscription Page config from Remnawave Panel by the user's "
|
||||
"subscription short UUID."
|
||||
),
|
||||
),
|
||||
SettingField(
|
||||
"SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED",
|
||||
"bool",
|
||||
"subscription_guides",
|
||||
"Enable admin JSON override",
|
||||
"Use the JSON field below instead of Remnawave Panel config. Disabled by default.",
|
||||
),
|
||||
SettingField(
|
||||
"SUBSCRIPTION_PAGE_CONFIG_PATH",
|
||||
"string",
|
||||
"subscription_guides",
|
||||
"Subscription Page config path",
|
||||
"Fallback path to a Remnawave Subscription Page v1 JSON config file.",
|
||||
placeholder="data/subpage-config/multiapp.json",
|
||||
),
|
||||
SettingField(
|
||||
"SUBSCRIPTION_PAGE_CONFIG_JSON",
|
||||
"json",
|
||||
"subscription_guides",
|
||||
"Subscription Page config JSON",
|
||||
(
|
||||
"Optional admin JSON override. It is applied only when the JSON override "
|
||||
"switch is enabled."
|
||||
),
|
||||
placeholder="{\n \"version\": \"1\"\n}",
|
||||
),
|
||||
# ─── Subscription periods & pricing ────────────────────────────
|
||||
SettingField("MONTH_1_ENABLED", "bool", "pricing", "Тариф 1 месяц"),
|
||||
SettingField("MONTH_3_ENABLED", "bool", "pricing", "Тариф 3 месяца"),
|
||||
@@ -454,6 +497,18 @@ def manifest_keys() -> List[str]:
|
||||
def coerce_value(field: SettingField, raw: Any) -> Any:
|
||||
"""Coerce a value coming from JSON to the type declared by the field."""
|
||||
|
||||
if field.type == "json":
|
||||
if raw is None:
|
||||
return ""
|
||||
text = raw if isinstance(raw, str) else str(raw)
|
||||
text = text.strip()
|
||||
if not text:
|
||||
return ""
|
||||
from config.subscription_guides_config import validate_subscription_guides_config_text
|
||||
|
||||
validate_subscription_guides_config_text(text)
|
||||
return text
|
||||
|
||||
if raw is None or (isinstance(raw, str) and raw.strip() == ""):
|
||||
return None
|
||||
|
||||
@@ -519,6 +574,7 @@ def manifest_payload() -> List[dict]:
|
||||
"notifications": 7,
|
||||
"support": 8,
|
||||
"devices": 9,
|
||||
"subscription_guides": 10,
|
||||
}
|
||||
items: List[dict] = []
|
||||
for field in aggregated_manifest():
|
||||
|
||||
@@ -11,6 +11,7 @@ from bot.app.web.webapp import (
|
||||
billing as _billing,
|
||||
common as _common,
|
||||
devices as _devices,
|
||||
guides as _guides,
|
||||
payloads as _payloads,
|
||||
routes as _routes,
|
||||
serializers as _serializers,
|
||||
@@ -27,6 +28,7 @@ _MODULES = (
|
||||
_serializers,
|
||||
_billing,
|
||||
_devices,
|
||||
_guides,
|
||||
_support,
|
||||
_routes,
|
||||
_application,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
# ruff: noqa: F401,F403,F405,I001
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
from .guides import warm_subscription_guides_config
|
||||
|
||||
|
||||
def create_subscription_webapp_application(
|
||||
@@ -24,6 +25,8 @@ def create_subscription_webapp_application(
|
||||
app["webapp_logo_cache"] = None
|
||||
app["webapp_logo_cache_lock"] = asyncio.Lock()
|
||||
app["webapp_settings_cache"] = {"ts": 0.0, "data": {}}
|
||||
app["subscription_guides_config_cache"] = {"fingerprint": None, "status": None}
|
||||
app["subscription_guides_config_lock"] = asyncio.Lock()
|
||||
app["webapp_rate_limit_buckets"] = {}
|
||||
app["webapp_rate_limit_lock"] = asyncio.Lock()
|
||||
|
||||
@@ -31,6 +34,7 @@ def create_subscription_webapp_application(
|
||||
await _ensure_shared_http_session()
|
||||
await _warm_webapp_logo_cache(app_obj)
|
||||
await _warm_webapp_animated_emoji_cache(app_obj)
|
||||
await warm_subscription_guides_config(app_obj)
|
||||
|
||||
async def _shutdown(app_obj: web.Application) -> None:
|
||||
await _close_shared_http_session()
|
||||
|
||||
@@ -2,13 +2,31 @@ from __future__ import annotations
|
||||
|
||||
from typing import Any, Awaitable, Callable, Optional
|
||||
|
||||
from bot.infra.redis import cache_delete, redis_key
|
||||
from bot.infra.redis import cache_delete, cache_delete_pattern, redis_key
|
||||
from bot.utils.ttl_cache import AsyncTTLCache
|
||||
from config.settings import Settings
|
||||
|
||||
_WEBAPP_USER_PAYLOAD_CACHES: dict[tuple[int, str, int], AsyncTTLCache] = {}
|
||||
|
||||
|
||||
def reset_webapp_settings_cache(app: Any) -> None:
|
||||
cache = app.get("webapp_settings_cache") if hasattr(app, "get") else None
|
||||
if isinstance(cache, dict):
|
||||
cache["ts"] = 0.0
|
||||
cache["data"] = {}
|
||||
|
||||
|
||||
def reset_subscription_guides_cache(app: Any) -> None:
|
||||
cache = app.get("subscription_guides_config_cache") if hasattr(app, "get") else None
|
||||
if isinstance(cache, dict):
|
||||
cache["fingerprint"] = None
|
||||
cache["status"] = None
|
||||
|
||||
|
||||
def _payload_namespaces(include_devices: bool = False) -> tuple[str, ...]:
|
||||
return ("me", "devices") if include_devices else ("me",)
|
||||
|
||||
|
||||
def _webapp_user_payload_cache(
|
||||
settings: Settings,
|
||||
namespace: str,
|
||||
@@ -55,6 +73,19 @@ def invalidate_local_webapp_user_payload(
|
||||
cache.invalidate(key)
|
||||
|
||||
|
||||
def invalidate_all_local_webapp_user_payloads(
|
||||
settings: Settings,
|
||||
*,
|
||||
include_devices: bool = False,
|
||||
) -> None:
|
||||
namespaces = set(_payload_namespaces(include_devices))
|
||||
for (settings_id, cache_namespace, _ttl), cache in tuple(
|
||||
_WEBAPP_USER_PAYLOAD_CACHES.items()
|
||||
):
|
||||
if settings_id == id(settings) and cache_namespace in namespaces:
|
||||
cache.invalidate()
|
||||
|
||||
|
||||
async def invalidate_webapp_user_caches(
|
||||
settings: Settings,
|
||||
*user_ids: Optional[int],
|
||||
@@ -79,3 +110,17 @@ async def invalidate_webapp_user_caches(
|
||||
invalidate_local_webapp_user_payload(settings, "devices", user_id)
|
||||
if keys:
|
||||
await cache_delete(settings, *keys)
|
||||
|
||||
|
||||
async def invalidate_all_webapp_user_payloads(
|
||||
settings: Settings,
|
||||
*,
|
||||
include_devices: bool = False,
|
||||
) -> None:
|
||||
invalidate_all_local_webapp_user_payloads(settings, include_devices=include_devices)
|
||||
for namespace in _payload_namespaces(include_devices):
|
||||
try:
|
||||
pattern = redis_key(settings, "cache", "webapp", namespace, "*")
|
||||
await cache_delete_pattern(settings, pattern)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
|
||||
from bot.app.web.webapp.cache_helpers import (
|
||||
invalidate_local_webapp_user_payload,
|
||||
invalidate_webapp_user_caches as _invalidate_user_payload_caches,
|
||||
)
|
||||
|
||||
|
||||
@@ -26,25 +26,7 @@ async def _invalidate_webapp_user_caches(
|
||||
*user_ids: Optional[int],
|
||||
include_devices: bool = False,
|
||||
) -> None:
|
||||
keys: List[str] = []
|
||||
seen: set[int] = set()
|
||||
for raw_user_id in user_ids:
|
||||
if raw_user_id is None:
|
||||
continue
|
||||
try:
|
||||
user_id = int(raw_user_id)
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
if user_id in seen:
|
||||
continue
|
||||
seen.add(user_id)
|
||||
keys.append(redis_key(settings, "cache", "webapp", "me", user_id))
|
||||
invalidate_local_webapp_user_payload(settings, "me", user_id)
|
||||
if include_devices:
|
||||
keys.append(redis_key(settings, "cache", "webapp", "devices", user_id))
|
||||
invalidate_local_webapp_user_payload(settings, "devices", user_id)
|
||||
if keys:
|
||||
await cache_delete(settings, *keys)
|
||||
await _invalidate_user_payload_caches(settings, *user_ids, include_devices=include_devices)
|
||||
|
||||
|
||||
def _validation_error_response(exc: ValidationError) -> web.Response:
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
# ruff: noqa: F401,F403,F405,I001
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
|
||||
from config.subscription_guides_config import (
|
||||
SubscriptionGuidesConfigError,
|
||||
subscription_guides_status,
|
||||
validate_panel_subscription_guides_config,
|
||||
)
|
||||
|
||||
PANEL_DEFAULT_SUBPAGE_CONFIG_UUID = "00000000-0000-0000-0000-000000000000"
|
||||
SUBSCRIPTION_GUIDES_CACHE_ERROR_TTL_SECONDS = 30
|
||||
|
||||
|
||||
async def warm_subscription_guides_config(app: web.Application) -> None:
|
||||
try:
|
||||
await _subscription_guides_status_shared(app)
|
||||
except Exception as exc:
|
||||
logger.warning("Failed to warm subscription guides config: %s", exc)
|
||||
|
||||
|
||||
async def subscription_guides_route(request: web.Request) -> web.Response:
|
||||
_require_user_id(request)
|
||||
status = await _subscription_guides_status_shared(request.app)
|
||||
payload = {
|
||||
"enabled": bool(status.get("enabled")),
|
||||
"config": status.get("config") if status.get("enabled") else None,
|
||||
"source": status.get("source"),
|
||||
}
|
||||
if status.get("error"):
|
||||
payload["error"] = status["error"]
|
||||
return web.json_response({"ok": True, **payload})
|
||||
|
||||
|
||||
async def public_subscription_guides_route(request: web.Request) -> web.Response:
|
||||
short_uuid = _normalize_short_uuid(request.match_info.get("short_uuid"))
|
||||
if not short_uuid:
|
||||
return web.json_response({"ok": False, "error": "invalid_short_uuid"}, status=404)
|
||||
|
||||
status = await _subscription_guides_status_shared(request.app)
|
||||
subscription = await _public_subscription_payload(request, short_uuid)
|
||||
payload = {
|
||||
"enabled": bool(status.get("enabled")),
|
||||
"config": status.get("config") if status.get("enabled") else None,
|
||||
"source": status.get("source"),
|
||||
"subscription": subscription,
|
||||
}
|
||||
if status.get("error"):
|
||||
payload["error"] = status["error"]
|
||||
return web.json_response({"ok": True, **payload})
|
||||
|
||||
|
||||
async def _subscription_guides_status_shared(app: web.Application) -> Dict[str, Any]:
|
||||
settings: Settings = app["settings"]
|
||||
cache = app.setdefault("subscription_guides_config_cache", {})
|
||||
lock: asyncio.Lock = app.setdefault("subscription_guides_config_lock", asyncio.Lock())
|
||||
fingerprint = _subscription_guides_settings_fingerprint(settings)
|
||||
now = time.monotonic()
|
||||
|
||||
cached = cache.get("status")
|
||||
if cached is not None and cache.get("fingerprint") == fingerprint:
|
||||
if cached.get("enabled") or now - float(cache.get("ts", 0.0)) < (
|
||||
SUBSCRIPTION_GUIDES_CACHE_ERROR_TTL_SECONDS
|
||||
):
|
||||
return cached
|
||||
|
||||
async with lock:
|
||||
cached = cache.get("status")
|
||||
if cached is not None and cache.get("fingerprint") == fingerprint:
|
||||
if cached.get("enabled") or now - float(cache.get("ts", 0.0)) < (
|
||||
SUBSCRIPTION_GUIDES_CACHE_ERROR_TTL_SECONDS
|
||||
):
|
||||
return cached
|
||||
|
||||
status = await _load_subscription_guides_status(app, settings)
|
||||
cache["fingerprint"] = fingerprint
|
||||
cache["status"] = status
|
||||
cache["ts"] = time.monotonic()
|
||||
return status
|
||||
|
||||
|
||||
async def _load_subscription_guides_status(
|
||||
app: web.Application,
|
||||
settings: Settings,
|
||||
) -> Dict[str, Any]:
|
||||
if not bool(getattr(settings, "SUBSCRIPTION_GUIDES_ENABLED", False)):
|
||||
return {"enabled": False, "config": None, "source": None, "error": None}
|
||||
|
||||
admin_json = str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON", "") or "").strip()
|
||||
json_override_enabled = bool(
|
||||
getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED", False)
|
||||
)
|
||||
if admin_json and json_override_enabled:
|
||||
return subscription_guides_status(settings)
|
||||
|
||||
if bool(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_PANEL_ENABLED", True)):
|
||||
panel_status = await _subscription_guides_status_from_panel_config(app, settings)
|
||||
if panel_status.get("enabled"):
|
||||
return panel_status
|
||||
|
||||
return subscription_guides_status(settings)
|
||||
|
||||
|
||||
async def _subscription_guides_status_from_panel_config(
|
||||
app: web.Application,
|
||||
settings: Settings,
|
||||
) -> Dict[str, Any]:
|
||||
panel_service = _panel_service_from_app(app)
|
||||
if panel_service is None:
|
||||
return {
|
||||
"enabled": False,
|
||||
"config": None,
|
||||
"source": "panel",
|
||||
"error": "Panel service is unavailable",
|
||||
}
|
||||
|
||||
try:
|
||||
config_uuid = str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_UUID", "") or "").strip()
|
||||
if not config_uuid:
|
||||
config_uuid = await _default_panel_subscription_page_config_uuid(panel_service)
|
||||
config_uuid = config_uuid or PANEL_DEFAULT_SUBPAGE_CONFIG_UUID
|
||||
detail = await panel_service.get_subscription_page_config_by_uuid(config_uuid)
|
||||
if detail is None and config_uuid != PANEL_DEFAULT_SUBPAGE_CONFIG_UUID:
|
||||
detail = await panel_service.get_subscription_page_config_by_uuid(
|
||||
PANEL_DEFAULT_SUBPAGE_CONFIG_UUID
|
||||
)
|
||||
if detail is None:
|
||||
raise SubscriptionGuidesConfigError(
|
||||
f"Panel subscription page config {config_uuid} is unavailable"
|
||||
)
|
||||
config = validate_panel_subscription_guides_config(detail)
|
||||
except (SubscriptionGuidesConfigError, Exception) as exc:
|
||||
logger.warning("Failed to load subscription guides config from Remnawave Panel: %s", exc)
|
||||
return {"enabled": False, "config": None, "source": "panel", "error": str(exc)}
|
||||
|
||||
return {"enabled": True, "config": config, "source": "panel", "error": None}
|
||||
|
||||
|
||||
async def _default_panel_subscription_page_config_uuid(panel_service: Any) -> str:
|
||||
get_list = getattr(panel_service, "get_subscription_page_config_list", None)
|
||||
if not callable(get_list):
|
||||
return ""
|
||||
payload = await get_list()
|
||||
configs = (payload or {}).get("configs")
|
||||
if not isinstance(configs, list):
|
||||
return ""
|
||||
|
||||
candidates: list[Dict[str, Any]] = [item for item in configs if isinstance(item, dict)]
|
||||
for item in candidates:
|
||||
uuid = str(item.get("uuid") or "").strip()
|
||||
if uuid == PANEL_DEFAULT_SUBPAGE_CONFIG_UUID:
|
||||
return uuid
|
||||
|
||||
candidates.sort(key=lambda item: int(item.get("viewPosition") or 0))
|
||||
for item in candidates:
|
||||
uuid = str(item.get("uuid") or "").strip()
|
||||
if uuid:
|
||||
return uuid
|
||||
return ""
|
||||
|
||||
|
||||
async def _public_subscription_payload(
|
||||
request: web.Request,
|
||||
short_uuid: str,
|
||||
) -> Dict[str, Any]:
|
||||
settings: Settings = request.app["settings"]
|
||||
panel_service = _panel_service_from_app(request.app)
|
||||
raw_link = ""
|
||||
username = ""
|
||||
resolved_short_uuid = short_uuid
|
||||
|
||||
async_session_factory: sessionmaker = request.app["async_session_factory"]
|
||||
async with async_session_factory() as session:
|
||||
local_sub = await subscription_dal.get_subscription_by_panel_subscription_uuid(
|
||||
session,
|
||||
short_uuid,
|
||||
)
|
||||
|
||||
if (
|
||||
local_sub
|
||||
and getattr(local_sub, "panel_user_uuid", None)
|
||||
and _local_subscription_is_publicly_active(local_sub)
|
||||
and panel_service
|
||||
):
|
||||
panel_user = await panel_service.get_user_by_uuid(local_sub.panel_user_uuid)
|
||||
if panel_user:
|
||||
raw_link = str(panel_user.get("subscriptionUrl") or "").strip()
|
||||
username = str(panel_user.get("username") or "").strip()
|
||||
resolved_short_uuid = str(panel_user.get("shortUuid") or short_uuid).strip()
|
||||
|
||||
display_link, connect_url = await prepare_config_links(settings, raw_link)
|
||||
return {
|
||||
"active": bool(display_link),
|
||||
"config_link": display_link,
|
||||
"connect_url": connect_url or display_link,
|
||||
"panel_short_uuid": resolved_short_uuid,
|
||||
"username": username,
|
||||
"share_url": _public_install_url(request, resolved_short_uuid),
|
||||
}
|
||||
|
||||
|
||||
def _panel_service_from_app(app: web.Application) -> Any:
|
||||
subscription_service: Optional[SubscriptionService] = app.get("subscription_service")
|
||||
panel_service = (
|
||||
getattr(subscription_service, "panel_service", None) if subscription_service else None
|
||||
)
|
||||
return panel_service or app.get("panel_service")
|
||||
|
||||
|
||||
def _subscription_guides_settings_fingerprint(settings: Settings) -> Tuple[Any, ...]:
|
||||
admin_json = str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON", "") or "")
|
||||
return (
|
||||
bool(getattr(settings, "SUBSCRIPTION_GUIDES_ENABLED", False)),
|
||||
bool(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_PANEL_ENABLED", True)),
|
||||
bool(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED", False)),
|
||||
str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_PATH", "") or ""),
|
||||
str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_UUID", "") or ""),
|
||||
hashlib.sha256(admin_json.encode("utf-8")).hexdigest(),
|
||||
str(getattr(settings, "PANEL_API_URL", "") or ""),
|
||||
bool(getattr(settings, "PANEL_API_KEY", "") or ""),
|
||||
)
|
||||
|
||||
|
||||
def _local_subscription_is_publicly_active(subscription: Any) -> bool:
|
||||
end_date = getattr(subscription, "end_date", None)
|
||||
if end_date and end_date.tzinfo is None:
|
||||
end_date = end_date.replace(tzinfo=timezone.utc)
|
||||
return bool(
|
||||
getattr(subscription, "is_active", False)
|
||||
and end_date
|
||||
and end_date > datetime.now(timezone.utc)
|
||||
)
|
||||
|
||||
|
||||
def _normalize_short_uuid(value: Any) -> str:
|
||||
short_uuid = str(value or "").strip()
|
||||
if not re.fullmatch(r"[A-Za-z0-9_-]{8,128}", short_uuid):
|
||||
return ""
|
||||
return short_uuid
|
||||
|
||||
|
||||
def _public_install_url(request: web.Request, short_uuid: str) -> str:
|
||||
settings: Settings = request.app["settings"]
|
||||
configured_base = str(getattr(settings, "SUBSCRIPTION_MINI_APP_URL", "") or "").strip()
|
||||
if configured_base:
|
||||
parts = urlsplit(configured_base)
|
||||
if parts.scheme and parts.netloc:
|
||||
base = urlunsplit((parts.scheme, parts.netloc, "", "", ""))
|
||||
else:
|
||||
base = configured_base.rstrip("/")
|
||||
else:
|
||||
host = (
|
||||
request.headers.get("X-Forwarded-Host")
|
||||
or request.headers.get("Host")
|
||||
or request.host
|
||||
)
|
||||
proto = request.headers.get("X-Forwarded-Proto") or request.scheme or "https"
|
||||
base = f"{proto}://{host}"
|
||||
return f"{base.rstrip('/')}/install/share/{quote(short_uuid)}"
|
||||
|
||||
|
||||
def _subscription_page_request_headers(request: web.Request) -> Dict[str, str]:
|
||||
headers = request.headers
|
||||
host = headers.get("X-Forwarded-Host") or headers.get("Host") or request.host
|
||||
proto = headers.get("X-Forwarded-Proto") or request.scheme or "https"
|
||||
user_agent = headers.get(
|
||||
"User-Agent",
|
||||
"Mozilla/5.0 AppleWebKit/537.36 (KHTML, like Gecko) Chrome Safari",
|
||||
)
|
||||
return {
|
||||
"host": host,
|
||||
"x-forwarded-host": host,
|
||||
"x-forwarded-proto": proto,
|
||||
"user-agent": user_agent,
|
||||
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||
"accept-language": headers.get("Accept-Language", "ru,en;q=0.9"),
|
||||
"sec-fetch-dest": "document",
|
||||
"sec-fetch-mode": "navigate",
|
||||
"sec-fetch-site": "none",
|
||||
"upgrade-insecure-requests": "1",
|
||||
}
|
||||
@@ -6,6 +6,8 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
|
||||
app.router.add_get("/", index_route)
|
||||
app.router.add_get("/login/password", index_route)
|
||||
app.router.add_get("/home", index_route)
|
||||
app.router.add_get("/install", index_route)
|
||||
app.router.add_get(r"/install/share/{short_uuid:[A-Za-z0-9_-]{8,128}}", index_route)
|
||||
app.router.add_get("/invite", index_route)
|
||||
app.router.add_get("/devices", index_route)
|
||||
app.router.add_get("/settings", index_route)
|
||||
@@ -60,6 +62,11 @@ def setup_subscription_webapp_routes(app: web.Application) -> None:
|
||||
app.router.add_get("/api/bootstrap", bootstrap_route)
|
||||
app.router.add_get("/api/i18n", i18n_route)
|
||||
app.router.add_get("/api/me", me_route)
|
||||
app.router.add_get("/api/subscription-guides", subscription_guides_route)
|
||||
app.router.add_get(
|
||||
r"/api/subscription-guides/public/{short_uuid:[A-Za-z0-9_-]{8,128}}",
|
||||
public_subscription_guides_route,
|
||||
)
|
||||
app.router.add_get("/api/account/avatar", account_avatar_route)
|
||||
app.router.add_post("/api/account/language", account_language_route)
|
||||
app.router.add_post("/api/account/email/request", account_email_request_route)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
# ruff: noqa: F401,F403,F405,I001
|
||||
from ._runtime import * # noqa: F403,F405
|
||||
|
||||
from config.subscription_guides_config import subscription_guides_available
|
||||
from config.webapp_themes_config import public_themes_catalog_payload
|
||||
|
||||
|
||||
@@ -82,7 +83,7 @@ async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, A
|
||||
"language_code": lang,
|
||||
"is_admin": is_admin,
|
||||
},
|
||||
"subscription": _serialize_subscription(settings, active, local_sub, lang),
|
||||
"subscription": _serialize_subscription(request, settings, active, local_sub, lang),
|
||||
"referral": {
|
||||
"code": referral_code,
|
||||
"bot_link": referral_link,
|
||||
@@ -132,6 +133,7 @@ async def _build_user_payload(request: web.Request, user_id: int) -> Dict[str, A
|
||||
"trial_traffic_limit_gb": float(settings.TRIAL_TRAFFIC_LIMIT_GB or 0),
|
||||
"trial_traffic_strategy": getattr(settings, "TRIAL_TRAFFIC_STRATEGY", "NO_RESET"),
|
||||
"subscription_purchase_description": settings.subscription_purchase_description(lang),
|
||||
"subscription_guides_enabled": subscription_guides_available(settings),
|
||||
"email_auth_enabled": settings.email_auth_configured,
|
||||
},
|
||||
}
|
||||
@@ -179,6 +181,7 @@ def _build_webapp_referral_link(
|
||||
|
||||
|
||||
def _serialize_subscription(
|
||||
request: web.Request,
|
||||
settings: Settings,
|
||||
active: Optional[Dict[str, Any]],
|
||||
local_sub: Optional[Any],
|
||||
@@ -192,6 +195,8 @@ def _serialize_subscription(
|
||||
"days_left": 0,
|
||||
"config_link": None,
|
||||
"connect_url": None,
|
||||
"panel_short_uuid": None,
|
||||
"install_share_url": None,
|
||||
}
|
||||
|
||||
end_date = active.get("end_date")
|
||||
@@ -231,6 +236,7 @@ def _serialize_subscription(
|
||||
can_topup_traffic = False
|
||||
can_topup_devices = False
|
||||
|
||||
panel_short_uuid = str(active.get("panel_short_uuid") or "").strip()
|
||||
return {
|
||||
"active": seconds_left > 0,
|
||||
"status": active.get("status_from_panel") or "UNKNOWN",
|
||||
@@ -240,6 +246,8 @@ def _serialize_subscription(
|
||||
"remaining_text": _format_remaining(seconds_left, lang),
|
||||
"config_link": active.get("config_link"),
|
||||
"connect_url": active.get("connect_button_url") or active.get("config_link"),
|
||||
"panel_short_uuid": panel_short_uuid or None,
|
||||
"install_share_url": _build_install_share_link(request, settings, panel_short_uuid),
|
||||
"traffic_limit": _format_bytes(active.get("traffic_limit_bytes"), zero_as_unlimited=True),
|
||||
"traffic_used": _format_bytes(active.get("traffic_used_bytes")),
|
||||
"traffic_limit_bytes": _coerce_int_or_none(active.get("traffic_limit_bytes")),
|
||||
@@ -284,6 +292,32 @@ def _serialize_subscription(
|
||||
}
|
||||
|
||||
|
||||
def _build_install_share_link(
|
||||
request: web.Request,
|
||||
settings: Settings,
|
||||
short_uuid: str,
|
||||
) -> Optional[str]:
|
||||
short_uuid = str(short_uuid or "").strip()
|
||||
if not short_uuid:
|
||||
return None
|
||||
configured_base = str(getattr(settings, "SUBSCRIPTION_MINI_APP_URL", "") or "").strip()
|
||||
if configured_base:
|
||||
parts = urlsplit(configured_base)
|
||||
if parts.scheme and parts.netloc:
|
||||
base = urlunsplit((parts.scheme, parts.netloc, "", "", ""))
|
||||
else:
|
||||
base = configured_base.rstrip("/")
|
||||
else:
|
||||
host = (
|
||||
request.headers.get("X-Forwarded-Host")
|
||||
or request.headers.get("Host")
|
||||
or request.host
|
||||
)
|
||||
proto = request.headers.get("X-Forwarded-Proto") or request.scheme or "https"
|
||||
base = f"{proto}://{host}"
|
||||
return f"{base.rstrip('/')}/install/share/{quote(short_uuid)}"
|
||||
|
||||
|
||||
def _serialize_plans(
|
||||
settings: Settings,
|
||||
lang: str,
|
||||
|
||||
@@ -597,6 +597,54 @@ class PanelApiService:
|
||||
return f"{base_sub_url}/{client_type.lower()}"
|
||||
return base_sub_url
|
||||
|
||||
async def get_subscription_page_config_by_short_uuid(
|
||||
self,
|
||||
short_uuid: str,
|
||||
request_headers: Optional[Dict[str, str]] = None,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
if not short_uuid:
|
||||
return None
|
||||
endpoint = f"/subscriptions/subpage-config/{short_uuid}"
|
||||
payload = {"requestHeaders": request_headers or {}}
|
||||
response_data = await self._request(
|
||||
"GET",
|
||||
endpoint,
|
||||
json=payload,
|
||||
log_full_response=False,
|
||||
)
|
||||
if response_data and not response_data.get("error"):
|
||||
return response_data.get("response", response_data)
|
||||
logging.error(
|
||||
f"Failed to get subscription page config for short UUID {short_uuid}. Response: {response_data}" # noqa: E501
|
||||
)
|
||||
return None
|
||||
|
||||
async def get_subscription_page_config_list(self) -> Optional[Dict[str, Any]]:
|
||||
endpoint = "/subscription-page-configs"
|
||||
response_data = await self._request("GET", endpoint, log_full_response=False)
|
||||
if response_data and not response_data.get("error"):
|
||||
return response_data.get("response", response_data)
|
||||
logging.error(
|
||||
f"Failed to get subscription page config list from panel. Response: {response_data}"
|
||||
)
|
||||
return None
|
||||
|
||||
async def get_subscription_page_config_by_uuid(
|
||||
self,
|
||||
config_uuid: str,
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
config_uuid = str(config_uuid or "").strip()
|
||||
if not config_uuid:
|
||||
return None
|
||||
endpoint = f"/subscription-page-configs/{config_uuid}"
|
||||
response_data = await self._request("GET", endpoint, log_full_response=False)
|
||||
if response_data and not response_data.get("error"):
|
||||
return response_data.get("response", response_data)
|
||||
logging.error(
|
||||
f"Failed to get subscription page config {config_uuid} from panel. Response: {response_data}" # noqa: E501
|
||||
)
|
||||
return None
|
||||
|
||||
async def get_user_devices(self, user_uuid: str) -> Optional[List[Dict[str, Any]]]:
|
||||
if self._devices_cache.ttl_seconds <= 0:
|
||||
return await self._get_user_devices_uncached(user_uuid)
|
||||
|
||||
@@ -763,6 +763,10 @@ class SubscriptionLifecycleMixin:
|
||||
|
||||
return {
|
||||
"user_id": panel_user_data.get("uuid"),
|
||||
"panel_subscription_uuid": panel_user_data.get("subscriptionUuid")
|
||||
or panel_user_data.get("shortUuid")
|
||||
or (local_active_sub.panel_subscription_uuid if local_active_sub else None),
|
||||
"panel_short_uuid": panel_user_data.get("shortUuid"),
|
||||
"end_date": panel_end_date,
|
||||
"status_from_panel": panel_user_data.get("status", "UNKNOWN").upper(),
|
||||
"config_link": display_link,
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -323,6 +323,26 @@ class Settings(BaseSettings):
|
||||
WEBAPP_FAVICON_USE_CUSTOM: bool = Field(default=False)
|
||||
WEBAPP_FAVICON_URL: Optional[str] = Field(default=None)
|
||||
WEBAPP_LOGO_FAVICON_URL: Optional[str] = Field(default=None)
|
||||
SUBSCRIPTION_GUIDES_ENABLED: bool = Field(
|
||||
default=True,
|
||||
description="Show embedded install instructions inside the subscription Mini App.",
|
||||
)
|
||||
SUBSCRIPTION_PAGE_CONFIG_PANEL_ENABLED: bool = Field(
|
||||
default=True,
|
||||
description="Use Remnawave Panel Subscription Page config for embedded guides when available.",
|
||||
)
|
||||
SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED: bool = Field(
|
||||
default=False,
|
||||
description="Enable admin JSON override for embedded guides config.",
|
||||
)
|
||||
SUBSCRIPTION_PAGE_CONFIG_PATH: str = Field(
|
||||
default="data/subpage-config/multiapp.json",
|
||||
description="Path to Remnawave Subscription Page v1 JSON config for embedded guides.",
|
||||
)
|
||||
SUBSCRIPTION_PAGE_CONFIG_JSON: str = Field(
|
||||
default="",
|
||||
description="Admin-provided Remnawave Subscription Page v1 JSON config override.",
|
||||
)
|
||||
WEBAPP_SESSION_SECRET: str = Field(default_factory=lambda: secrets.token_urlsafe(32))
|
||||
WEBHOOK_SECRET_TOKEN: str = Field(default_factory=lambda: secrets.token_urlsafe(32))
|
||||
WEBAPP_SESSION_TTL_SECONDS: int = Field(default=24 * 60 * 60)
|
||||
|
||||
@@ -0,0 +1,669 @@
|
||||
"""Loader and validator for Remnawave Subscription Page v1 configs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Iterable, Mapping, Optional, Tuple
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
|
||||
class SubscriptionGuidesConfigError(ValueError):
|
||||
"""Raised when the embedded subscription guides config is invalid."""
|
||||
|
||||
|
||||
APP_ROOT = Path(__file__).resolve().parents[2]
|
||||
DEFAULT_CONFIG_PATH = "data/subpage-config/multiapp.json"
|
||||
DEFAULT_CONFIG_BUNDLED_PATH = (
|
||||
Path(__file__).resolve().parent / "defaults" / "subscription_page_multiapp.json"
|
||||
)
|
||||
ALLOWED_LOCALES = {
|
||||
"az",
|
||||
"be",
|
||||
"de",
|
||||
"en",
|
||||
"es",
|
||||
"fa",
|
||||
"fr",
|
||||
"hi",
|
||||
"id",
|
||||
"ja",
|
||||
"pl",
|
||||
"pt",
|
||||
"ru",
|
||||
"th",
|
||||
"tk",
|
||||
"tr",
|
||||
"uk",
|
||||
"uz",
|
||||
"vi",
|
||||
"zh",
|
||||
}
|
||||
ALLOWED_PLATFORMS = {
|
||||
"android",
|
||||
"androidTV",
|
||||
"appleTV",
|
||||
"ios",
|
||||
"linux",
|
||||
"macos",
|
||||
"windows",
|
||||
}
|
||||
ALLOWED_BUTTON_TYPES = {"copyButton", "external", "subscriptionLink"}
|
||||
BASE_TRANSLATION_KEYS = (
|
||||
"active",
|
||||
"bandwidth",
|
||||
"connectionKeysHeader",
|
||||
"copyLink",
|
||||
"expired",
|
||||
"expires",
|
||||
"expiresIn",
|
||||
"getLink",
|
||||
"inactive",
|
||||
"indefinitely",
|
||||
"installationGuideHeader",
|
||||
"linkCopied",
|
||||
"linkCopiedToClipboard",
|
||||
"name",
|
||||
"scanQrCode",
|
||||
"scanQrCodeDescription",
|
||||
"scanToImport",
|
||||
"status",
|
||||
"unknown",
|
||||
)
|
||||
ALLOWED_SVG_COLORS = {
|
||||
"red",
|
||||
"orange",
|
||||
"amber",
|
||||
"yellow",
|
||||
"lime",
|
||||
"green",
|
||||
"emerald",
|
||||
"teal",
|
||||
"cyan",
|
||||
"sky",
|
||||
"blue",
|
||||
"indigo",
|
||||
"violet",
|
||||
"purple",
|
||||
"fuchsia",
|
||||
"pink",
|
||||
"rose",
|
||||
"slate",
|
||||
"gray",
|
||||
"zinc",
|
||||
"neutral",
|
||||
"stone",
|
||||
}
|
||||
UI_SUBSCRIPTION_INFO_TYPES = {"cards", "collapsed", "expanded", "hidden"}
|
||||
UI_INSTALLATION_GUIDE_TYPES = {"accordion", "cards", "minimal", "timeline"}
|
||||
SVG_KEY_RE = re.compile(r"^[A-Za-z]+$")
|
||||
HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{3,8}$")
|
||||
CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
|
||||
UNSAFE_SVG_RE = re.compile(
|
||||
r"(<\s*/?\s*(?:script|foreignObject|iframe|object|embed|image|use|style|a)\b)"
|
||||
r"|(\son[a-z]+\s*=)"
|
||||
r"|(javascript\s*:)"
|
||||
r"|(data\s*:)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CONFIG_CACHE: Dict[Tuple[str, str], Dict[str, Any]] = {}
|
||||
PANEL_CONFIG_KEYS = (
|
||||
"config",
|
||||
"subscriptionPageConfig",
|
||||
"subpageConfig",
|
||||
"subPageConfig",
|
||||
"pageConfig",
|
||||
)
|
||||
PANEL_WRAPPER_KEYS = ("response", "data", "result", *PANEL_CONFIG_KEYS)
|
||||
|
||||
|
||||
def validate_subscription_guides_config_text(raw: str) -> Dict[str, Any]:
|
||||
"""Parse and validate a v1 subscription guides config JSON string."""
|
||||
|
||||
try:
|
||||
payload = json.loads(raw)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise SubscriptionGuidesConfigError(f"Invalid JSON: {exc.msg}") from exc
|
||||
return validate_subscription_guides_config(payload)
|
||||
|
||||
|
||||
def default_subscription_guides_config_text() -> str:
|
||||
try:
|
||||
return DEFAULT_CONFIG_BUNDLED_PATH.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
raise SubscriptionGuidesConfigError(
|
||||
f"Bundled default config is unavailable: {exc}"
|
||||
) from exc
|
||||
|
||||
|
||||
def resolve_subscription_guides_config_path(settings: Any) -> Path:
|
||||
configured_path = str(
|
||||
getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_PATH", DEFAULT_CONFIG_PATH)
|
||||
or DEFAULT_CONFIG_PATH
|
||||
).strip()
|
||||
if not configured_path:
|
||||
raise SubscriptionGuidesConfigError("SUBSCRIPTION_PAGE_CONFIG_PATH is empty")
|
||||
path = Path(configured_path)
|
||||
if not path.is_absolute():
|
||||
path = APP_ROOT / path
|
||||
return path
|
||||
|
||||
|
||||
def ensure_subscription_guides_config_file(settings: Any) -> Path:
|
||||
path = resolve_subscription_guides_config_path(settings)
|
||||
if not path.exists():
|
||||
raise SubscriptionGuidesConfigError(f"Config file does not exist: {path}")
|
||||
return path
|
||||
|
||||
|
||||
def subscription_guides_admin_config_json(settings: Any) -> Tuple[str, str]:
|
||||
admin_json = str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON", "") or "").strip()
|
||||
if admin_json:
|
||||
return admin_json, "admin_json"
|
||||
return "", "empty"
|
||||
|
||||
|
||||
def load_subscription_guides_config(settings: Any) -> Tuple[Dict[str, Any], str]:
|
||||
"""Load the enabled guides config from admin JSON or a configured file path."""
|
||||
|
||||
source, raw = _read_config_source(settings)
|
||||
digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||
cache_key = (source, digest)
|
||||
cached = _CONFIG_CACHE.get(cache_key)
|
||||
if cached is not None:
|
||||
return copy.deepcopy(cached), source
|
||||
config = validate_subscription_guides_config_text(raw)
|
||||
_CONFIG_CACHE[cache_key] = copy.deepcopy(config)
|
||||
return config, source
|
||||
|
||||
|
||||
def subscription_guides_status(settings: Any) -> Dict[str, Any]:
|
||||
"""Return a safe status payload for user-facing guide availability checks."""
|
||||
|
||||
if not bool(getattr(settings, "SUBSCRIPTION_GUIDES_ENABLED", False)):
|
||||
return {"enabled": False, "config": None, "source": None, "error": None}
|
||||
try:
|
||||
config, source = load_subscription_guides_config(settings)
|
||||
except SubscriptionGuidesConfigError as exc:
|
||||
return {"enabled": False, "config": None, "source": None, "error": str(exc)}
|
||||
return {"enabled": True, "config": config, "source": source, "error": None}
|
||||
|
||||
|
||||
def subscription_guides_available(settings: Any) -> bool:
|
||||
if not bool(getattr(settings, "SUBSCRIPTION_GUIDES_ENABLED", False)):
|
||||
return False
|
||||
admin_json = str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON", "") or "").strip()
|
||||
if (
|
||||
not (
|
||||
admin_json
|
||||
and bool(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED", False))
|
||||
)
|
||||
and bool(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_PANEL_ENABLED", True))
|
||||
and getattr(settings, "PANEL_API_URL", None)
|
||||
and getattr(settings, "PANEL_API_KEY", None)
|
||||
):
|
||||
return True
|
||||
status = subscription_guides_status(settings)
|
||||
return bool(status.get("enabled") and status.get("config"))
|
||||
|
||||
|
||||
def extract_subscription_guides_config_from_panel(payload: Any) -> Any:
|
||||
"""Extract Subscription Page v1 config from flexible Panel API response shapes."""
|
||||
|
||||
return _extract_config_candidate(payload, set())
|
||||
|
||||
|
||||
def validate_panel_subscription_guides_config(
|
||||
payload: Any,
|
||||
*,
|
||||
allow_default_when_missing: bool = False,
|
||||
) -> Dict[str, Any]:
|
||||
config = extract_subscription_guides_config_from_panel(payload)
|
||||
if config is None:
|
||||
if allow_default_when_missing and panel_subscription_page_allowed(payload):
|
||||
default_text = default_subscription_guides_config_text()
|
||||
return validate_subscription_guides_config_text(default_text)
|
||||
raise SubscriptionGuidesConfigError("Panel response does not contain a v1 config")
|
||||
return validate_subscription_guides_config(config)
|
||||
|
||||
|
||||
def panel_subscription_page_allowed(payload: Any) -> bool:
|
||||
candidate = _find_panel_response_object(payload, set())
|
||||
return bool(candidate and candidate.get("webpageAllowed") is True)
|
||||
|
||||
|
||||
def validate_subscription_guides_config(payload: Any) -> Dict[str, Any]:
|
||||
if not isinstance(payload, Mapping):
|
||||
raise SubscriptionGuidesConfigError("Config root must be an object")
|
||||
if payload.get("version") != "1":
|
||||
raise SubscriptionGuidesConfigError(
|
||||
"Only Subscription Page config version '1' is supported"
|
||||
)
|
||||
|
||||
locales = _validate_locales(payload.get("locales"))
|
||||
svg_library = _validate_svg_library(payload.get("svgLibrary"))
|
||||
branding = _validate_branding(payload.get("brandingSettings"))
|
||||
ui_config = _validate_ui_config(payload.get("uiConfig"))
|
||||
base_settings = _validate_base_settings(payload.get("baseSettings"))
|
||||
base_translations = _validate_base_translations(payload.get("baseTranslations"), locales)
|
||||
platforms = _validate_platforms(payload.get("platforms"), locales, svg_library)
|
||||
|
||||
return {
|
||||
"version": "1",
|
||||
"locales": locales,
|
||||
"brandingSettings": branding,
|
||||
"uiConfig": ui_config,
|
||||
"baseSettings": base_settings,
|
||||
"baseTranslations": base_translations,
|
||||
"svgLibrary": svg_library,
|
||||
"platforms": platforms,
|
||||
}
|
||||
|
||||
|
||||
def _read_config_source(settings: Any) -> Tuple[str, str]:
|
||||
admin_json = str(getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON", "") or "").strip()
|
||||
json_override_enabled = bool(
|
||||
getattr(settings, "SUBSCRIPTION_PAGE_CONFIG_JSON_OVERRIDE_ENABLED", False)
|
||||
)
|
||||
if admin_json and json_override_enabled:
|
||||
return "admin_json", admin_json
|
||||
|
||||
path = ensure_subscription_guides_config_file(settings)
|
||||
try:
|
||||
return "file", path.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
raise SubscriptionGuidesConfigError(f"Failed to read config file: {exc}") from exc
|
||||
|
||||
|
||||
def _extract_config_candidate(value: Any, seen: set[int]) -> Any:
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
if not text.startswith(("{", "[")):
|
||||
return None
|
||||
try:
|
||||
return _extract_config_candidate(json.loads(text), seen)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise SubscriptionGuidesConfigError(f"Invalid JSON in Panel config: {exc.msg}") from exc
|
||||
|
||||
if not isinstance(value, Mapping):
|
||||
return None
|
||||
|
||||
value_id = id(value)
|
||||
if value_id in seen:
|
||||
return None
|
||||
seen.add(value_id)
|
||||
|
||||
if _looks_like_v1_config(value):
|
||||
return value
|
||||
|
||||
for key in PANEL_WRAPPER_KEYS:
|
||||
if key not in value:
|
||||
continue
|
||||
candidate = _extract_config_candidate(value.get(key), seen)
|
||||
if candidate is not None:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _looks_like_v1_config(value: Mapping[str, Any]) -> bool:
|
||||
return (
|
||||
value.get("version") == "1"
|
||||
and "locales" in value
|
||||
and "svgLibrary" in value
|
||||
and "platforms" in value
|
||||
)
|
||||
|
||||
|
||||
def _find_panel_response_object(value: Any, seen: set[int]) -> Optional[Mapping[str, Any]]:
|
||||
if not isinstance(value, Mapping):
|
||||
return None
|
||||
|
||||
value_id = id(value)
|
||||
if value_id in seen:
|
||||
return None
|
||||
seen.add(value_id)
|
||||
|
||||
if "webpageAllowed" in value:
|
||||
return value
|
||||
|
||||
for key in ("response", "data", "result"):
|
||||
candidate = _find_panel_response_object(value.get(key), seen)
|
||||
if candidate is not None:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _validate_locales(value: Any) -> list[str]:
|
||||
if not isinstance(value, list) or not value:
|
||||
raise SubscriptionGuidesConfigError("locales must be a non-empty array")
|
||||
locales: list[str] = []
|
||||
for index, item in enumerate(value):
|
||||
locale = str(item or "").strip()
|
||||
if locale not in ALLOWED_LOCALES:
|
||||
raise SubscriptionGuidesConfigError(f"Unsupported locale at locales[{index}]: {locale}")
|
||||
if locale not in locales:
|
||||
locales.append(locale)
|
||||
return locales
|
||||
|
||||
|
||||
def _validate_branding(value: Any) -> Dict[str, str]:
|
||||
data = _require_object(value, "brandingSettings")
|
||||
result = {
|
||||
"title": _require_text(data, "title", "brandingSettings.title"),
|
||||
"logoUrl": _require_text(data, "logoUrl", "brandingSettings.logoUrl"),
|
||||
"supportUrl": _require_text(data, "supportUrl", "brandingSettings.supportUrl"),
|
||||
}
|
||||
_assert_http_url(result["logoUrl"], "brandingSettings.logoUrl")
|
||||
_assert_http_url(result["supportUrl"], "brandingSettings.supportUrl")
|
||||
return result
|
||||
|
||||
|
||||
def _validate_ui_config(value: Any) -> Dict[str, str]:
|
||||
data = _require_object(value, "uiConfig")
|
||||
subscription_info = _require_text(
|
||||
data,
|
||||
"subscriptionInfoBlockType",
|
||||
"uiConfig.subscriptionInfoBlockType",
|
||||
)
|
||||
installation_guides = _require_text(
|
||||
data,
|
||||
"installationGuidesBlockType",
|
||||
"uiConfig.installationGuidesBlockType",
|
||||
)
|
||||
if subscription_info not in UI_SUBSCRIPTION_INFO_TYPES:
|
||||
raise SubscriptionGuidesConfigError(
|
||||
f"Unsupported uiConfig.subscriptionInfoBlockType: {subscription_info}"
|
||||
)
|
||||
if installation_guides not in UI_INSTALLATION_GUIDE_TYPES:
|
||||
raise SubscriptionGuidesConfigError(
|
||||
f"Unsupported uiConfig.installationGuidesBlockType: {installation_guides}"
|
||||
)
|
||||
return {
|
||||
"subscriptionInfoBlockType": subscription_info,
|
||||
"installationGuidesBlockType": installation_guides,
|
||||
}
|
||||
|
||||
|
||||
def _validate_base_settings(value: Any) -> Dict[str, Any]:
|
||||
data = value if isinstance(value, Mapping) else {}
|
||||
return {
|
||||
"metaTitle": _optional_text(data, "metaTitle") or "Subscription",
|
||||
"metaDescription": _optional_text(data, "metaDescription") or "Subscription",
|
||||
"showConnectionKeys": bool(data.get("showConnectionKeys", False)),
|
||||
"hideGetLinkButton": bool(data.get("hideGetLinkButton", False)),
|
||||
}
|
||||
|
||||
|
||||
def _validate_base_translations(value: Any, locales: Iterable[str]) -> Dict[str, Dict[str, str]]:
|
||||
data = _require_object(value, "baseTranslations")
|
||||
result: Dict[str, Dict[str, str]] = {}
|
||||
for key in BASE_TRANSLATION_KEYS:
|
||||
result[key] = _validate_locale_strings(
|
||||
data.get(key),
|
||||
locales,
|
||||
f"baseTranslations.{key}",
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def _validate_svg_library(value: Any) -> Dict[str, str]:
|
||||
data = _require_object(value, "svgLibrary")
|
||||
if not data:
|
||||
raise SubscriptionGuidesConfigError("svgLibrary must not be empty")
|
||||
result: Dict[str, str] = {}
|
||||
for key, raw_svg in data.items():
|
||||
svg_key = str(key or "").strip()
|
||||
if not SVG_KEY_RE.fullmatch(svg_key):
|
||||
raise SubscriptionGuidesConfigError(f"Invalid svgLibrary key: {svg_key}")
|
||||
result[svg_key] = _sanitize_svg(raw_svg, f"svgLibrary.{svg_key}")
|
||||
return result
|
||||
|
||||
|
||||
def _validate_platforms(
|
||||
value: Any,
|
||||
locales: Iterable[str],
|
||||
svg_library: Mapping[str, str],
|
||||
) -> Dict[str, Dict[str, Any]]:
|
||||
data = _require_object(value, "platforms")
|
||||
if not data:
|
||||
raise SubscriptionGuidesConfigError("platforms must not be empty")
|
||||
result: Dict[str, Dict[str, Any]] = {}
|
||||
for platform_key, raw_platform in data.items():
|
||||
key = str(platform_key or "").strip()
|
||||
if key not in ALLOWED_PLATFORMS:
|
||||
raise SubscriptionGuidesConfigError(f"Unsupported platform: {key}")
|
||||
platform = _require_object(raw_platform, f"platforms.{key}")
|
||||
icon_key = _validate_svg_icon_key(
|
||||
platform.get("svgIconKey"),
|
||||
svg_library,
|
||||
f"platforms.{key}.svgIconKey",
|
||||
)
|
||||
apps = _validate_apps(platform.get("apps"), locales, svg_library, f"platforms.{key}.apps")
|
||||
result[key] = {
|
||||
"displayName": _validate_localized_or_text(
|
||||
platform.get("displayName"),
|
||||
locales,
|
||||
f"platforms.{key}.displayName",
|
||||
),
|
||||
"svgIconKey": icon_key,
|
||||
"apps": apps,
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def _validate_apps(
|
||||
value: Any,
|
||||
locales: Iterable[str],
|
||||
svg_library: Mapping[str, str],
|
||||
path: str,
|
||||
) -> list[Dict[str, Any]]:
|
||||
if not isinstance(value, list) or not value:
|
||||
raise SubscriptionGuidesConfigError(f"{path} must be a non-empty array")
|
||||
apps: list[Dict[str, Any]] = []
|
||||
for index, raw_app in enumerate(value):
|
||||
app_path = f"{path}[{index}]"
|
||||
app = _require_object(raw_app, app_path)
|
||||
name = _require_text(app, "name", f"{app_path}.name")
|
||||
if len(name) < 2:
|
||||
raise SubscriptionGuidesConfigError(f"{app_path}.name must contain at least 2 chars")
|
||||
icon_key = _optional_svg_icon_key(
|
||||
app.get("svgIconKey"),
|
||||
svg_library,
|
||||
f"{app_path}.svgIconKey",
|
||||
)
|
||||
apps.append(
|
||||
{
|
||||
"name": name,
|
||||
"svgIconKey": icon_key,
|
||||
"featured": bool(app.get("featured", False)),
|
||||
"blocks": _validate_blocks(
|
||||
app.get("blocks"),
|
||||
locales,
|
||||
svg_library,
|
||||
f"{app_path}.blocks",
|
||||
),
|
||||
}
|
||||
)
|
||||
return apps
|
||||
|
||||
|
||||
def _validate_blocks(
|
||||
value: Any,
|
||||
locales: Iterable[str],
|
||||
svg_library: Mapping[str, str],
|
||||
path: str,
|
||||
) -> list[Dict[str, Any]]:
|
||||
if not isinstance(value, list) or not value:
|
||||
raise SubscriptionGuidesConfigError(f"{path} must be a non-empty array")
|
||||
blocks: list[Dict[str, Any]] = []
|
||||
for index, raw_block in enumerate(value):
|
||||
block_path = f"{path}[{index}]"
|
||||
block = _require_object(raw_block, block_path)
|
||||
color = _optional_text(block, "svgIconColor")
|
||||
if color and color not in ALLOWED_SVG_COLORS and not HEX_COLOR_RE.fullmatch(color):
|
||||
raise SubscriptionGuidesConfigError(f"{block_path}.svgIconColor is invalid")
|
||||
blocks.append(
|
||||
{
|
||||
"svgIconKey": _validate_svg_icon_key(
|
||||
block.get("svgIconKey"),
|
||||
svg_library,
|
||||
f"{block_path}.svgIconKey",
|
||||
),
|
||||
"svgIconColor": color or "",
|
||||
"title": _validate_locale_strings(
|
||||
block.get("title"),
|
||||
locales,
|
||||
f"{block_path}.title",
|
||||
),
|
||||
"description": _validate_locale_strings(
|
||||
block.get("description"),
|
||||
locales,
|
||||
f"{block_path}.description",
|
||||
),
|
||||
"buttons": _validate_buttons(
|
||||
block.get("buttons"),
|
||||
locales,
|
||||
svg_library,
|
||||
f"{block_path}.buttons",
|
||||
),
|
||||
}
|
||||
)
|
||||
return blocks
|
||||
|
||||
|
||||
def _validate_buttons(
|
||||
value: Any,
|
||||
locales: Iterable[str],
|
||||
svg_library: Mapping[str, str],
|
||||
path: str,
|
||||
) -> list[Dict[str, Any]]:
|
||||
if value is None:
|
||||
return []
|
||||
if not isinstance(value, list):
|
||||
raise SubscriptionGuidesConfigError(f"{path} must be an array")
|
||||
buttons: list[Dict[str, Any]] = []
|
||||
for index, raw_button in enumerate(value):
|
||||
button_path = f"{path}[{index}]"
|
||||
button = _require_object(raw_button, button_path)
|
||||
button_type = _require_text(button, "type", f"{button_path}.type")
|
||||
if button_type not in ALLOWED_BUTTON_TYPES:
|
||||
raise SubscriptionGuidesConfigError(
|
||||
f"Unsupported button type at {button_path}: {button_type}"
|
||||
)
|
||||
link = _require_text(button, "link", f"{button_path}.link")
|
||||
_validate_button_link(link, button_type, f"{button_path}.link")
|
||||
buttons.append(
|
||||
{
|
||||
"type": button_type,
|
||||
"link": link,
|
||||
"text": _validate_locale_strings(
|
||||
button.get("text"),
|
||||
locales,
|
||||
f"{button_path}.text",
|
||||
),
|
||||
"svgIconKey": _validate_svg_icon_key(
|
||||
button.get("svgIconKey"),
|
||||
svg_library,
|
||||
f"{button_path}.svgIconKey",
|
||||
),
|
||||
}
|
||||
)
|
||||
return buttons
|
||||
|
||||
|
||||
def _validate_locale_strings(value: Any, locales: Iterable[str], path: str) -> Dict[str, str]:
|
||||
data = _require_object(value, path)
|
||||
result: Dict[str, str] = {}
|
||||
for locale in locales:
|
||||
text = data.get(locale)
|
||||
if not isinstance(text, str) or not text.strip():
|
||||
raise SubscriptionGuidesConfigError(f"{path}.{locale} is required")
|
||||
result[locale] = text.strip()
|
||||
return result
|
||||
|
||||
|
||||
def _validate_localized_or_text(
|
||||
value: Any,
|
||||
locales: Iterable[str],
|
||||
path: str,
|
||||
) -> str | Dict[str, str]:
|
||||
if isinstance(value, str):
|
||||
text = value.strip()
|
||||
if text:
|
||||
return text
|
||||
return _validate_locale_strings(value, locales, path)
|
||||
|
||||
|
||||
def _validate_svg_icon_key(value: Any, svg_library: Mapping[str, str], path: str) -> str:
|
||||
key = _string_value(value)
|
||||
if not key:
|
||||
raise SubscriptionGuidesConfigError(f"{path} is required")
|
||||
if key not in svg_library:
|
||||
raise SubscriptionGuidesConfigError(f"{path} references missing svgLibrary key: {key}")
|
||||
return key
|
||||
|
||||
|
||||
def _optional_svg_icon_key(value: Any, svg_library: Mapping[str, str], path: str) -> Optional[str]:
|
||||
key = _string_value(value)
|
||||
if not key:
|
||||
return None
|
||||
if key not in svg_library:
|
||||
raise SubscriptionGuidesConfigError(f"{path} references missing svgLibrary key: {key}")
|
||||
return key
|
||||
|
||||
|
||||
def _validate_button_link(link: str, _button_type: str, path: str) -> None:
|
||||
_assert_safe_link(link, path)
|
||||
|
||||
|
||||
def _assert_safe_link(value: str, path: str) -> None:
|
||||
if CONTROL_CHARS_RE.search(value):
|
||||
raise SubscriptionGuidesConfigError(f"{path} contains control characters")
|
||||
lower = value.strip().lower()
|
||||
if lower.startswith(("javascript:", "data:", "vbscript:")):
|
||||
raise SubscriptionGuidesConfigError(f"{path} uses an unsafe URL scheme")
|
||||
|
||||
|
||||
def _assert_http_url(value: str, path: str) -> None:
|
||||
_assert_safe_link(value, path)
|
||||
parts = urlsplit(value)
|
||||
if parts.scheme not in {"http", "https"} or not parts.netloc:
|
||||
raise SubscriptionGuidesConfigError(f"{path} must be an http(s) URL")
|
||||
|
||||
|
||||
def _sanitize_svg(value: Any, path: str) -> str:
|
||||
svg = _string_value(value)
|
||||
if not svg:
|
||||
raise SubscriptionGuidesConfigError(f"{path} is required")
|
||||
trimmed = svg.strip()
|
||||
if not trimmed.lower().startswith("<svg"):
|
||||
raise SubscriptionGuidesConfigError(f"{path} must be an SVG document")
|
||||
if UNSAFE_SVG_RE.search(trimmed):
|
||||
raise SubscriptionGuidesConfigError(f"{path} contains unsafe SVG markup")
|
||||
return trimmed
|
||||
|
||||
|
||||
def _require_object(value: Any, path: str) -> Mapping[str, Any]:
|
||||
if not isinstance(value, Mapping):
|
||||
raise SubscriptionGuidesConfigError(f"{path} must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def _require_text(data: Mapping[str, Any], key: str, path: str) -> str:
|
||||
value = _string_value(data.get(key))
|
||||
if not value:
|
||||
raise SubscriptionGuidesConfigError(f"{path} is required")
|
||||
return value
|
||||
|
||||
|
||||
def _optional_text(data: Mapping[str, Any], key: str) -> str:
|
||||
return _string_value(data.get(key))
|
||||
|
||||
|
||||
def _string_value(value: Any) -> str:
|
||||
if not isinstance(value, str):
|
||||
return ""
|
||||
return value.strip()
|
||||
Reference in New Issue
Block a user