diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index eecaa46d..f21b80ee 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -30,10 +30,24 @@ use its own authentication and generic reverse-proxy contract. `app.app_factory.create_app()` is the single production construction path for the Flask application. Importing `app.web` defines the core HTTP surface and its accessor facade, but does not construct an application. Factory creation -has a fixed order: configure Flask, establish the durable signing key, attach -runtime state, synchronize authentication state, register core routes, -register core blueprints, load module contributions and templates, install -base-path handling, then apply reverse-proxy handling as the outer WSGI layer. +first resolves one immutable registration plan containing core routes, core +blueprints, built-in modules, and enabled community modules. Blueprint +callbacks are preflighted on an isolated application, and URL/config ownership +collisions are validated before the target application or process-wide module +catalogs are changed. The validated plan then applies its HTTP, catalog, and +template contributions once, followed by base-path handling and reverse-proxy +handling as the outer WSGI layer. + +`app.registration` is the only productive Flask registrar. It owns blueprint +and direct-rule application, endpoint/blueprint/route-method collision checks, +and the canonical registration manifest. The manifest contains stable public +identifiers only; its SHA-256 fingerprint is logged at startup and stored in +the application extensions. Filesystem locations, callables, and configuration +or secret names and values are excluded. `app.module_registry` owns manifest +discovery, `app.module_config_registry` owns configuration-schema preflight, +`app.module_contributions` owns contribution-specific resolution and preflight, +and `app.module_loader` orchestrates ownership and rejection policy behind its +compatible public facade. Each application owns a typed `DocsightRuntime` at `app.extensions["docsight"]`. It contains the configuration manager, storage, @@ -515,7 +529,20 @@ Installed non-theme modules are discovered by `ModuleLoader` and persisted throu Module manifests can declare module-owned config defaults through the top-level `config` object. Normal module config remains plain local configuration. A community manifest may opt specific declared string-default keys into write-only secret handling with `config_secrets`; the value must be a list of unique strings, every listed key must exist in that manifest's `config`, and each corresponding default must itself be a string so encrypted values never enter scalar coercion paths. The pure-stdlib validator in `app/manifest_contract.py` is the authoritative manifest capability contract used by the runtime loader and can also validate external catalogs without importing Flask. -Before any enabled module loads, `ModuleLoader` reserves secret ownership from every discovered community manifest, including disabled modules. Core secret, hash-backed, private, and core configuration keys cannot be claimed. A secret key must also be exclusive to its declaring module's `config`; claims that overlap another module's plain or secret config fail every affected module and grant no owner. The installer performs the same ownership evaluation against installed modules and rejects a conflicting package before it can be persisted. Valid module secrets are encrypted by `ConfigManager`, masked in Settings responses, and exposed through the community config proxy only to their owning module; other module secrets and all core protected values are removed. +Before any enabled module loads, `ModuleLoader` resolves secret ownership from every discovered community manifest, including disabled modules, as fail-closed safety metadata. Core secret, hash-backed, private, and core configuration keys cannot be claimed. A secret key must also be exclusive to its declaring module's `config`; claims that overlap another module's plain or secret config fail every affected module and grant no owner. Normal community configuration keys also require one unambiguous owner. No ownership or classification is applied until the complete registration plan validates. The installer performs the same ownership evaluation against installed modules and rejects a conflicting package before it can be persisted. Valid module secrets are encrypted by `ConfigManager`, masked in Settings responses, and exposed through the community config proxy only to their owning module; other module secrets and all core protected values are removed. + +Module source order is stable: the explicit built-in directory registry, the +built-in threshold and theme registries, then configured community search paths +and lexicographically sorted directories within each path. Duplicate community +IDs retain the first source for manifest-v1 compatibility and skip later +sources deterministically. Disabled and rejected modules remain visible as +metadata but apply no routes, static mounts, templates, config defaults, i18n, +active themes, thresholds, collectors, or publishers. As a metadata-only +exception, a disabled community theme's declared `theme.json` is resolved and +validated through the same safe contribution path so Appearance can render its +preview. Invalid or unsafe preview data rejects that module, and preview data +never creates a registration-plan contribution. A built-in planning failure +aborts application construction; a community failure rejects only that module. Settings receives the active module-secret and saved-secret key sets from the server, so installed templates from before the explicit field-marker contract still preserve masked values safely during a coordinated catalog rollout. Current templates also represent saved secrets locally with empty password inputs plus explicit `data-config-secret` and `data-saved-secret` metadata. Untouched saved fields submit the standard mask, which `ConfigManager` treats as preserve-existing, while edited fields submit the replacement value. Saved plaintext is never rendered into HTML. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 58aef2de..1534686c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -105,6 +105,10 @@ Open `http://localhost:8765` to access the setup wizard. ``` app/ app_factory.py - Deterministic Flask application construction + registration.py - Registration planning, collision checks, and application + module_registry.py - Stable module manifest discovery and metadata + module_config_registry.py - Module configuration ownership preflight + module_contributions.py - Module filesystem/import/JSON contribution preflight main.py - Entrypoint, ThreadPoolExecutor polling loop runtime.py - Typed per-application runtime state and locks web.py - Core routes, filters, auth, and runtime accessors @@ -163,6 +167,25 @@ We prefer new languages to be contributed by people who actually use the tool in DOCSight supports community modules that extend functionality without modifying core code. Modules can add API endpoints, data collectors, settings panels, dashboard tabs, and more. +Route contributions export a Flask `Blueprint`; they must not call +`app.register_blueprint()`, `app.add_url_rule()`, or `Blueprint.register()` +themselves. Routes and decorators recorded on the exported blueprint are the +module contribution consumed by the registrar. +Deferred Blueprint record callbacks run once on an isolated application during +preflight and again during real registration. Community callbacks must +therefore be idempotent and free of filesystem, network, process, or other +external side effects. +The shared application registration contract preflights the complete core and +module plan before applying it. Endpoint names, blueprint names, route/method +pairs, module IDs, static mounts, and configuration ownership must therefore be +globally unique. Add registration-contract tests when changing this surface, +including a failure case that proves the target application and process-wide +catalogs remain unchanged. +Every explicitly declared contribution must resolve and validate during that +preflight. An unresolved route, static directory, template, catalog, class, or +JSON contribution rejects the module's complete plan; no partial contribution +is registered. + Server-side module code should import the established accessors it needs from `app.web`, such as `get_config_manager()`, `get_storage()`, `get_state()`, or `get_module_loader()`. These accessors resolve the active application's typed diff --git a/app/app_factory.py b/app/app_factory.py index 1c333dfb..d4e491b1 100644 --- a/app/app_factory.py +++ b/app/app_factory.py @@ -9,13 +9,20 @@ from typing import Any from flask import Flask -from jinja2 import ChoiceLoader, FileSystemLoader from werkzeug.middleware.proxy_fix import ProxyFix from . import web from .base_path import configure_base_path -from .blueprints import register_blueprints from .module_loader import ModuleLoader +from .registration import ( + RegistrationPlan, + build_core_plan, + canonical_manifest, + existing_rules, + manifest_fingerprint, + register_plan, + validate_plan, +) from .runtime import ( AuthStateStore, DocsightRuntime, @@ -54,19 +61,6 @@ def build(app: Flask): return build -def install_module_template_loader(app: Flask, module_loader) -> None: - """Add enabled module template directories to this app's Jinja loader.""" - if module_loader is None: - return - loaders = [app.jinja_loader] - for module in module_loader.get_enabled_modules(): - template_dir = os.path.join(module.path, "templates") - if os.path.isdir(template_dir): - loaders.append(FileSystemLoader(template_dir)) - if len(loaders) > 1: - app.jinja_loader = ChoiceLoader(loaders) - - def apply_reverse_proxy(app: Flask, environ: Mapping[str, str]) -> None: """Install trusted reverse-proxy handling as the outer WSGI layer.""" reverse_proxy = environ.get("REVERSE_PROXY", "").strip() @@ -98,15 +92,30 @@ def create_app( ) -> Flask: """Create one fully isolated DOCSight application in a fixed order. - Construction configures Flask, attaches runtime state, initializes auth, - registers core and blueprint routes, loads modules and templates, then - installs base-path and reverse-proxy middleware. + Construction resolves and validates the complete core/module plan before + configuring the target app or applying process catalogs. It then attaches + runtime state, applies the complete registration once, and wraps the + base-path and reverse-proxy middleware. """ if config_manager is None: raise TypeError("config_manager is required") env = os.environ if environ is None else environ + core_plan = build_core_plan() + planning_app = Flask("app.registration_planning", static_folder=None) + planning_app.config["TESTING"] = testing + planning_app.extensions["docsight_registration_deferred"] = True + planning_app.extensions["docsight_registration_base_plan"] = core_plan + loader = module_loader_factory(planning_app) if module_loader_factory else None + module_plan = getattr(loader, "registration_plan", RegistrationPlan()) + complete_plan = core_plan.combined(module_plan) + app = Flask("app.web", template_folder="templates") + validate_plan( + complete_plan, + existing=existing_rules(app), + existing_blueprints=tuple(app.blueprints), + ) app.config.update( SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", @@ -137,12 +146,15 @@ def create_app( ) attach_runtime(app, runtime) web.bootstrap_auth_state(app, runtime) - web.register_core_routes(app) - register_blueprints(app) - - loader = module_loader_factory(app) if module_loader_factory else None + register_plan(app, complete_plan) + web.install_core_template_hooks(app) runtime.module_loader = loader - install_module_template_loader(app, loader) configure_base_path(app, env) apply_reverse_proxy(app, env) + manifest = canonical_manifest(app, loader) + fingerprint = manifest_fingerprint(manifest) + app.extensions["docsight_module_loader"] = loader + app.extensions["docsight_registration_manifest"] = manifest + app.extensions["docsight_registration_fingerprint"] = fingerprint + LOG.info("registration manifest sha256=%s", fingerprint) return app diff --git a/app/blueprints/__init__.py b/app/blueprints/__init__.py index 3ac19f6f..73b8c4f5 100644 --- a/app/blueprints/__init__.py +++ b/app/blueprints/__init__.py @@ -1,7 +1,7 @@ -"""Flask Blueprint registration.""" +"""Stable providers for DOCSight's core Flask blueprints.""" -def register_blueprints(app): +def core_blueprints(): from .config_bp import config_bp from .polling_bp import polling_bp from .data_bp import data_bp @@ -13,13 +13,15 @@ def register_blueprints(app): from .segment_bp import segment_bp from .smart_capture_bp import smart_capture_bp - app.register_blueprint(config_bp) - app.register_blueprint(polling_bp) - app.register_blueprint(data_bp) - app.register_blueprint(analysis_bp) - app.register_blueprint(events_bp) - app.register_blueprint(modules_bp) - app.register_blueprint(metrics_bp) - app.register_blueprint(notices_bp) - app.register_blueprint(segment_bp) - app.register_blueprint(smart_capture_bp) + return ( + config_bp, + polling_bp, + data_bp, + analysis_bp, + events_bp, + modules_bp, + metrics_bp, + notices_bp, + segment_bp, + smart_capture_bp, + ) diff --git a/app/module_config_registry.py b/app/module_config_registry.py new file mode 100644 index 00000000..68243585 --- /dev/null +++ b/app/module_config_registry.py @@ -0,0 +1,107 @@ +"""Preflight and application of module-owned configuration schema.""" + +from __future__ import annotations + +from collections import defaultdict +from typing import Any + +from . import config as _cfg +from .module_registry import ModuleInfo + + +def register_module_config( + config_defaults: dict[str, Any], + module_id: str | None = None, + builtin: bool = False, + config_secrets: list[str] | None = None, + config_private: list[str] | None = None, +) -> set[str]: + """Apply prevalidated defaults and classification to the process schema.""" + secret_keys = set(config_secrets or []) + private_keys = set(config_private or []) if builtin else set() + registered: set[str] = set() + for key, value in config_defaults.items(): + if key in secret_keys and not builtin and _cfg.MODULE_SECRET_OWNERS.get(key) != module_id: + continue + if key in _cfg.DEFAULTS: + if key in private_keys: + _cfg.PRIVATE_KEYS.add(key) + continue + if not builtin and key in (_cfg.SECRET_KEYS | _cfg.PRIVATE_KEYS | _cfg.HASH_KEYS): + continue + _cfg.DEFAULTS[key] = value + registered.add(key) + if key in secret_keys and builtin: + _cfg.SECRET_KEYS.add(key) + if key in private_keys: + _cfg.PRIVATE_KEYS.add(key) + if _cfg.is_secret_key(key): + continue + if isinstance(value, bool): + _cfg.BOOL_KEYS.add(key) + elif isinstance(value, int): + _cfg.INT_KEYS.add(key) + return registered + + +def evaluate_module_secret_ownership( + modules: list[ModuleInfo], +) -> tuple[set[str], dict[str, str], dict[str, str]]: + """Return reserved keys, valid owners, and redacted ownership errors.""" + protected = set(_cfg.CORE_CONFIG_KEYS) | _cfg.SECRET_KEYS | _cfg.HASH_KEYS | _cfg.PRIVATE_KEYS + for module in modules: + if module.builtin: + protected.update(module.config) + secret_claims: dict[str, list[ModuleInfo]] = defaultdict(list) + config_claims: dict[str, list[ModuleInfo]] = defaultdict(list) + for module in modules: + if module.builtin: + continue + for key in module.config: + config_claims[key].append(module) + for key in module.config_secrets: + secret_claims[key].append(module) + reserved = {key for key in secret_claims if key not in protected} + owners: dict[str, str] = {} + errors: dict[str, str] = {} + for key, claimants in secret_claims.items(): + if key in protected: + for module in claimants: + errors[module.id] = "Module secret declaration conflicts with protected configuration" + continue + users = config_claims.get(key, []) + if len(claimants) != 1 or len(users) != 1 or users[0].id != claimants[0].id: + for module in [*claimants, *users]: + errors[module.id] = "Module secret ownership conflict" + continue + owners[key] = claimants[0].id + return reserved, owners, errors + + +def reserve_module_secrets(modules: list[ModuleInfo]) -> None: + """Compatibility adapter for applying preflighted secret ownership.""" + reserved, owners, errors = evaluate_module_secret_ownership(modules) + for module in modules: + if module.id in errors: + module.error = errors[module.id] + _cfg.set_module_secret_registry(reserved, owners) + + +def evaluate_module_config_ownership(modules: list[ModuleInfo]) -> dict[str, str]: + """Return redacted errors for ambiguous community config ownership.""" + protected = set(_cfg.CORE_CONFIG_KEYS) + protected.update(key for module in modules if module.builtin for key in module.config) + claims: dict[str, list[ModuleInfo]] = defaultdict(list) + errors: dict[str, str] = {} + for module in modules: + if module.builtin or not module.enabled: + continue + for key in module.config: + claims[key].append(module) + if key in protected: + errors[module.id] = "Module config ownership conflicts with protected configuration" + for claimants in claims.values(): + if len(claimants) > 1: + for module in claimants: + errors[module.id] = "Module config ownership conflict" + return errors diff --git a/app/module_contributions.py b/app/module_contributions.py new file mode 100644 index 00000000..dc070fe4 --- /dev/null +++ b/app/module_contributions.py @@ -0,0 +1,452 @@ +"""Resolve and preflight complete module contribution sets.""" + +import importlib +import importlib.util +import json +import logging +import os +import sys +from typing import Any + +from flask import abort, send_file, url_for + +from .builtin_modules import BUILTIN_PYTHON_CONTRIBUTIONS +from .module_registry import ManifestError, ModuleInfo +from .path_safety import safe_manifest_ref, safe_manifest_subpath +from .registration import ( + ModuleContribution, + PlannedRule, + RegistrationError, + RegistrationPlan, + apply_module_i18n, + apply_plan, + probe_blueprint, +) + + +log = logging.getLogger("docsis.modules") + +_PROTECTED_ROUTES = { + "/", "/login", "/logout", "/setup", "/settings", "/health", "/sw.js", +} +_PROTECTED_API_PREFIXES = ( + "/api/config", "/api/data", "/api/tokens", "/api/demo", + "/api/poll", "/api/status", "/api/history", "/api/events", "/api/trends", + "/api/export", "/api/correlation", "/api/modules/", "/api/themes/", +) +REQUIRED_THRESHOLD_SECTIONS = {"downstream_power", "upstream_power", "snr"} +REQUIRED_THEME_SECTIONS = {"dark", "light"} + + +def resolve_module_i18n(i18n_dir: str) -> dict[str, dict[str, Any]]: + """Read module translations without mutating the process catalog.""" + if not os.path.isdir(i18n_dir): + return {} + catalogs: dict[str, dict[str, Any]] = {} + for fname in sorted(os.listdir(i18n_dir)): + if not fname.endswith(".json") or fname == "template.json": + continue + fpath = os.path.join(i18n_dir, fname) + try: + with open(fpath, "r", encoding="utf-8") as handle: + data = json.load(handle) + except (json.JSONDecodeError, OSError) as exc: + raise ManifestError("Invalid module translation catalog") from exc + if not isinstance(data, dict): + raise ManifestError("Module translation catalog must be an object") + catalogs[fname[:-5]] = data + return catalogs + + +def merge_module_i18n(module_id: str, i18n_dir: str) -> None: + """Compatibility adapter that resolves then applies one i18n contribution.""" + apply_module_i18n(module_id, resolve_module_i18n(i18n_dir)) + + +def _load_symbol(spec: str, module_id: str): + """Import a trusted built-in Python contribution by module path.""" + if ":" not in spec: + log.warning("Built-in module '%s': invalid Python contribution spec", module_id) + return None + module_name, attr_name = spec.rsplit(":", 1) + try: + module = importlib.import_module(module_name) + except Exception: + log.error("Built-in module '%s': Python contribution import failed", module_id) + return None + value = getattr(module, attr_name, None) + if value is None: + log.warning("Built-in module '%s': Python contribution symbol not found", module_id) + return value + + +def attach_builtin_python_contributions(mod: ModuleInfo) -> None: + """Attach statically registered Python entry points for a built-in module.""" + specs = BUILTIN_PYTHON_CONTRIBUTIONS.get(mod.id) + for key, attr_name in ( + ("collector", "collector_class"), + ("publisher", "publisher_class"), + ): + if key not in mod.contributes or getattr(mod, attr_name) is not None: + continue + spec = getattr(specs, key, None) if specs else None + if not spec: + raise ManifestError(f"Built-in module '{mod.id}' missing static {key} registration") + symbol = _load_symbol(spec, mod.id) + if not isinstance(symbol, type): + raise ManifestError( + f"Built-in module '{mod.id}' failed to import static {key} registration" + ) + setattr(mod, attr_name, symbol) + + +def resolve_module_routes( + module_id: str, + module_path: str, + routes_file: str, + *, + builtin: bool = False, +) -> RegistrationPlan: + """Import and preflight a module Blueprint without target mutation.""" + routes_path = safe_manifest_ref(module_path, routes_file) + if not os.path.isfile(routes_path): + raise ManifestError("Routes contribution file not found") + dir_name = os.path.basename(module_path) + if builtin: + mod_name = f"app.modules.{dir_name}.{os.path.splitext(os.path.basename(routes_file))[0]}" + try: + mod = importlib.import_module(mod_name) + except Exception as exc: + raise ManifestError("Routes contribution import failed") from exc + else: + mod_name = f"community_modules.{dir_name}.routes" + try: + spec = importlib.util.spec_from_file_location(mod_name, routes_path) + if spec is None or spec.loader is None: + raise ManifestError("Routes contribution import failed") + mod = importlib.util.module_from_spec(spec) + sys.modules[mod_name] = mod + spec.loader.exec_module(mod) + except Exception as exc: + raise ManifestError("Routes contribution import failed") from exc + blueprint = getattr(mod, "bp", None) or getattr(mod, "blueprint", None) + if blueprint is None: + raise ManifestError("Routes contribution does not export a Blueprint") + planned = probe_blueprint(blueprint, source=f"module:{module_id}") + if not builtin: + blocked = [] + for rule in planned.rules: + path = rule.rule + if path in _PROTECTED_ROUTES: + blocked.append(path) + continue + for prefix in _PROTECTED_API_PREFIXES: + if path.startswith(prefix): + if prefix == "/api/modules/" and path.startswith( + f"/api/modules/{module_id}/" + ): + own_action = path.rstrip("/") + if own_action not in { + f"/api/modules/{module_id}/enable", + f"/api/modules/{module_id}/disable", + }: + continue + blocked.append(path) + break + if blocked: + raise RegistrationError( + f"Module {module_id} has protected route conflicts: " + + ", ".join(sorted(blocked)) + ) + return RegistrationPlan(blueprints=(planned,)) + + +def load_module_routes( + app, + module_id: str, + module_path: str, + routes_file: str, + *, + builtin: bool = False, +) -> None: + """Compatibility adapter for registering one independently tested module.""" + try: + plan = resolve_module_routes(module_id, module_path, routes_file, builtin=builtin) + apply_plan(app, plan) + except RegistrationError: + raise + except (ManifestError, OSError): + log.warning("Module '%s': routes contribution skipped", module_id) + + +def _load_module_class(module_id: str, module_path: str, spec: str, kind: str): + """Load a contributed class from a module-owned Python file.""" + if ":" not in spec: + log.warning("Module '%s': invalid %s contribution spec", module_id, kind) + return None + filename, class_name = spec.rsplit(":", 1) + file_path = safe_manifest_ref(module_path, filename) + if not os.path.isfile(file_path): + log.warning("Module '%s': %s contribution file not found", module_id, kind) + return None + dir_name = os.path.basename(module_path) + mod_name = f"app.modules.{dir_name}.{kind}" + try: + im_spec = importlib.util.spec_from_file_location(mod_name, file_path) + if im_spec is None or im_spec.loader is None: + log.warning("Module '%s': %s contribution import unavailable", module_id, kind) + return None + mod = importlib.util.module_from_spec(im_spec) + sys.modules[mod_name] = mod + im_spec.loader.exec_module(mod) + except Exception: + log.error("Module '%s': %s contribution import failed", module_id, kind) + return None + cls = getattr(mod, class_name, None) + if cls is None: + log.warning("Module '%s': %s contribution class not found", module_id, kind) + return None + log.info("Module '%s': loaded %s contribution", module_id, kind) + return cls + + +def load_module_collector(module_id: str, module_path: str, spec: str): + """Load a Collector class, returning None on resolution failure.""" + return _load_module_class(module_id, module_path, spec, "collector") + + +def load_module_publisher(module_id: str, module_path: str, spec: str): + """Load a Publisher class, returning None on resolution failure.""" + return _load_module_class(module_id, module_path, spec, "publisher") + + +def module_static_endpoint(module_id: str) -> str: + """Return the stable Flask endpoint name for a module's static files.""" + return f"module_static_{module_id}" + + +def module_static_url(module_id: str, filename: str, **values: Any) -> str: + """Build a module-static URL using the registered endpoint contract.""" + return url_for(module_static_endpoint(module_id), filename=filename, **values) + + +def plan_module_static( + module_id: str, module_path: str, static_subdir: str +) -> RegistrationPlan: + """Resolve a safe module-static mount without target mutation.""" + static_dir = safe_manifest_subpath(module_path, static_subdir.rstrip("/")) + if not os.path.isdir(static_dir): + return RegistrationPlan() + static_root = os.path.realpath(static_dir) + static_prefix = static_root + os.sep + route = f"/modules/{module_id}/static/" + + def serve_static(filename, _root=static_root, _prefix=static_prefix): + try: + candidate = os.path.realpath(os.path.join(_root, filename)) + except (OSError, TypeError, ValueError): + return abort(404) + if candidate.startswith(_prefix): + if not os.path.isfile(candidate): + return abort(404) + return send_file(candidate) + return abort(404) + + endpoint = module_static_endpoint(module_id) + return RegistrationPlan(rules=(PlannedRule( + route, endpoint, ("GET",), f"module-static:{module_id}", serve_static, + ),)) + + +def setup_module_static(app, module_id: str, module_path: str, static_subdir: str) -> None: + """Compatibility adapter for mounting one independently tested module.""" + apply_plan(app, plan_module_static(module_id, module_path, static_subdir)) + + +def setup_module_templates( + module_id: str, module_path: str, contributes: dict[str, str] +) -> dict[str, str]: + """Resolve declared template files for Jinja includes.""" + resolved = {} + for key in {"tab", "card", "settings"}: + rel_path = contributes.get(key) + if not rel_path: + continue + abs_path = safe_manifest_subpath(module_path, rel_path) + if os.path.isfile(abs_path): + resolved[key] = os.path.basename(abs_path) + log.debug("Module '%s': resolved %s template contribution", module_id, key) + else: + log.warning("Module '%s': %s template contribution not found", module_id, key) + return resolved + + +def validate_thresholds(data: dict[str, object]) -> None: + """Validate a threshold contribution.""" + missing = REQUIRED_THRESHOLD_SECTIONS - set(data.keys()) + if missing: + raise ManifestError(f"Missing required threshold sections: {', '.join(sorted(missing))}") + for section in REQUIRED_THRESHOLD_SECTIONS: + block = data[section] + if not isinstance(block, dict): + raise ManifestError(f"Threshold section '{section}' must be a dict") + if "_default" not in block: + raise ManifestError(f"Threshold section '{section}' missing '_default' key") + + +def validate_theme(data: dict[str, object]) -> None: + """Validate a theme contribution.""" + if not isinstance(data, dict): + raise ManifestError("Theme contribution must be an object") + missing = REQUIRED_THEME_SECTIONS - set(data.keys()) + if missing: + raise ManifestError(f"Missing required theme sections: {', '.join(sorted(missing))}") + for section in REQUIRED_THEME_SECTIONS: + block = data[section] + if not isinstance(block, dict): + raise ManifestError(f"Theme section '{section}' must be a dict") + if not block: + raise ManifestError(f"Theme section '{section}' is empty") + for key, value in block.items(): + if not isinstance(value, str): + raise ManifestError( + f"Theme property '{key}' in '{section}' must be a string, " + f"got {type(value).__name__}" + ) + + +def _read_json_contribution( + module_path: str, reference: str, kind: str, validator +) -> dict[str, object]: + try: + path = safe_manifest_ref(module_path, reference) + except ValueError as exc: + raise ManifestError(f"{kind} contribution reference is unsafe") from exc + if not os.path.isfile(path): + raise ManifestError(f"{kind} contribution file not found") + try: + with open(path, "r", encoding="utf-8") as handle: + data = json.load(handle) + validator(data) + return data + except (json.JSONDecodeError, OSError, ManifestError) as exc: + raise ManifestError(f"{kind} contribution is invalid") from exc + + +def _redacted_resolution(kind: str, resolver): + try: + return resolver() + except ValueError as exc: + raise ManifestError(f"{kind} contribution reference is unsafe") from exc + + +def resolve_module_contribution( + mod: ModuleInfo, +) -> tuple[ModuleContribution, RegistrationPlan]: + """Resolve and validate one complete contribution set without target mutation.""" + contributes = mod.contributes + http = RegistrationPlan() + collector_class = publisher_class = None + if mod.builtin: + attach_builtin_python_contributions(mod) + collector_class, publisher_class = mod.collector_class, mod.publisher_class + static_subdir = contributes.get("static", "static/").rstrip("/") + static_dir = _redacted_resolution( + "static", lambda: safe_manifest_subpath(mod.path, static_subdir) + ) + if "i18n" in contributes: + i18n_dir = _redacted_resolution( + "i18n", + lambda: safe_manifest_subpath(mod.path, contributes["i18n"].rstrip("/")), + ) + if not os.path.isdir(i18n_dir): + raise ManifestError("i18n contribution directory not found") + i18n_catalogs = resolve_module_i18n(i18n_dir) + else: + i18n_catalogs = {} + if "routes" in contributes: + http = _redacted_resolution( + "routes", + lambda: resolve_module_routes( + mod.id, mod.path, contributes["routes"], builtin=mod.builtin + ), + ) + if "static" in contributes and not os.path.isdir(static_dir): + raise ManifestError("static contribution directory not found") + if os.path.isdir(static_dir): + http = http.combined(plan_module_static(mod.id, mod.path, static_subdir)) + has_css = os.path.isfile(os.path.join(static_dir, "style.css")) + has_js = os.path.isfile(os.path.join(static_dir, "main.js")) + else: + has_css = has_js = False + template_paths = _redacted_resolution( + "template", lambda: setup_module_templates(mod.id, mod.path, contributes) + ) + declared_templates = { + key for key in ("tab", "card", "settings") if key in contributes + } + if declared_templates != set(template_paths): + missing_kind = sorted(declared_templates - set(template_paths))[0] + raise ManifestError(f"{missing_kind} template contribution file not found") + if "collector" in contributes and not mod.builtin: + collector_class = _redacted_resolution( + "collector", + lambda: load_module_collector( + mod.id, mod.path, contributes["collector"] + ), + ) + if not isinstance(collector_class, type): + raise ManifestError("collector contribution could not be resolved") + if "publisher" in contributes and not mod.builtin: + publisher_class = _redacted_resolution( + "publisher", + lambda: load_module_publisher( + mod.id, mod.path, contributes["publisher"] + ), + ) + if not isinstance(publisher_class, type): + raise ManifestError("publisher contribution could not be resolved") + if "thresholds" in contributes: + thresholds_data = mod.thresholds_data + if thresholds_data is None: + thresholds_data = _read_json_contribution( + mod.path, contributes["thresholds"], "thresholds", validate_thresholds + ) + else: + validate_thresholds(thresholds_data) + else: + thresholds_data = None + if "theme" in contributes: + theme_data = mod.theme_data + if theme_data is None: + theme_data = _read_json_contribution( + mod.path, contributes["theme"], "theme", validate_theme + ) + else: + validate_theme(theme_data) + else: + theme_data = None + template_dir = os.path.join(mod.path, "templates") + contribution = ModuleContribution( + module_id=mod.id, + source=("builtin:" if mod.builtin else "community:") + mod.id, + version=mod.version, + builtin=mod.builtin, + info=mod, + config=tuple(sorted(mod.config.items())), + secret_keys=tuple(sorted(mod.config_secrets)), + private_keys=tuple(sorted(mod.config_private if mod.builtin else ())), + i18n_catalogs=tuple( + (lang, tuple(sorted(strings.items()))) + for lang, strings in sorted(i18n_catalogs.items()) + ), + template_paths=tuple(sorted(template_paths.items())), + template_dir=template_dir if os.path.isdir(template_dir) else None, + collector_class=collector_class, + publisher_class=publisher_class, + thresholds_data=thresholds_data, + theme_data=theme_data, + has_css=has_css, + has_js=has_js, + ) + return contribution, http diff --git a/app/module_loader.py b/app/module_loader.py index e50f1642..ab2b79b4 100644 --- a/app/module_loader.py +++ b/app/module_loader.py @@ -1,1009 +1,225 @@ -"""Module loader: discovers, validates, and loads DOCSight modules.""" +"""Module discovery orchestration and compatibility facade.""" -import importlib -import importlib.util -import json import logging -import os -import sys -from collections import defaultdict -from copy import deepcopy -from dataclasses import dataclass, field -from typing import Any, cast +import os # Compatibility: callers patch the shared filesystem module here. -from flask import abort, send_file, url_for - -from app import analyzer as _analyzer -from app import config as _cfg +# Public imports below intentionally preserve the legacy module-loader facade. +from app import module_registry as _module_registry from app.builtin_modules import BUILTIN_MODULE_DIRS, BUILTIN_PYTHON_CONTRIBUTIONS -from app.i18n import _TRANSLATIONS -from app.manifest_contract import ( - ID_PATTERN, - REQUIRED_FIELDS, - VALID_CONTRIBUTES, - VALID_TYPES, - validate_manifest_contract, +from app.manifest_contract import ID_PATTERN, REQUIRED_FIELDS, VALID_CONTRIBUTES, VALID_TYPES +from app.module_config_registry import ( + evaluate_module_config_ownership, + evaluate_module_secret_ownership, + register_module_config, + reserve_module_secrets, +) +from app.module_contributions import ( + REQUIRED_THEME_SECTIONS, REQUIRED_THRESHOLD_SECTIONS, + _PROTECTED_API_PREFIXES, _PROTECTED_ROUTES, + _load_module_class, _load_symbol, + _read_json_contribution, _redacted_resolution, + attach_builtin_python_contributions, load_module_collector, + load_module_publisher, load_module_routes, merge_module_i18n, + module_static_endpoint, module_static_url, plan_module_static, + resolve_module_contribution, resolve_module_i18n, resolve_module_routes, + setup_module_static, setup_module_templates, validate_theme, + validate_thresholds, +) +from app.module_registry import ( + ManifestError, + ModuleInfo, + ModuleRegistryError, + discover_modules, + validate_manifest, ) from app.path_safety import safe_manifest_ref, safe_manifest_subpath +from app.registration import ( + ModuleContribution, PlannedBlueprint, PlannedRule, RegistrationError, + RegistrationPlan, apply_module_i18n, apply_plan, existing_rules, + probe_blueprint, register_plan, validate_plan, +) from app.theme_registry import BUILTIN_THEMES from app.threshold_profiles import BUILTIN_THRESHOLD_PROFILES -log = logging.getLogger("docsis.modules") - -class ManifestError(Exception): - """Raised when a manifest.json is invalid.""" - - -@dataclass -class ModuleInfo: - """Validated module metadata from manifest.json.""" - id: str - name: str - description: str - version: str - author: str - min_app_version: str - type: str - contributes: dict[str, str] - path: str - builtin: bool = False - homepage: str = "" - license: str = "" - config: dict[str, Any] = field(default_factory=dict) - config_secrets: list[str] = field(default_factory=list) - config_private: list[str] = field(default_factory=list) - menu: dict[str, Any] = field(default_factory=dict) - enabled: bool = True - error: str | None = None - template_paths: dict[str, str] = field(default_factory=dict) - collector_class: type | None = None - publisher_class: type | None = None - hints: dict[str, object] = field(default_factory=dict) - thresholds_data: dict[str, object] | None = None - theme_data: dict[str, object] | None = None - has_css: bool = False - has_js: bool = False - - -def validate_manifest(raw: dict[str, Any], module_path: str, *, builtin: bool | None = None) -> ModuleInfo: - """Validate a raw manifest dict and return a ModuleInfo. - - Raises ManifestError if the manifest is invalid. - """ - # Detect builtin unless the caller already knows the module source. - if builtin is None: - norm = os.path.normpath(module_path).replace("\\", "/") - builtin = "/app/modules/" in norm or "\\app\\modules\\" in os.path.normpath(module_path) - - errors = validate_manifest_contract(raw, builtin=builtin) - if errors: - raise ManifestError(errors[0]) - - mod_id = raw["id"] - mod_type = raw["type"] - contributes = raw["contributes"] - config = raw.get("config", {}) - config_secrets = raw.get("config_secrets", []) - config_private = raw.get("configPrivate", []) - - return ModuleInfo( - id=mod_id, - name=raw["name"], - description=raw["description"], - version=raw["version"], - author=raw["author"], - min_app_version=raw["minAppVersion"], - type=mod_type, - contributes=contributes, - path=module_path, - builtin=builtin, - homepage=raw.get("homepage", ""), - license=raw.get("license", ""), - config=config, - config_secrets=config_secrets, - config_private=config_private, - menu={**{"order": 999}, **raw.get("menu", {})}, - hints=raw.get("hints", {}), - ) - - -def discover_modules( - search_paths: list[str] | None = None, - disabled_ids: set[str] | None = None, - known_ids: set[str] | None = None, -) -> list[ModuleInfo]: - """Scan directories for module manifest.json files. - - Args: - search_paths: List of directories to scan. Each directory is expected - to contain subdirectories, each with a manifest.json. - disabled_ids: Set of module IDs that should be marked as disabled. - known_ids: Module IDs that have already been registered by a - higher-priority source. Matching manifests are skipped as - duplicates. - - Returns: - List of validated ModuleInfo objects. Invalid manifests are logged - and skipped -- they never raise exceptions. - """ - if search_paths is None: - search_paths = [] - if disabled_ids is None: - disabled_ids = set() - - modules: list[ModuleInfo] = [] - seen_ids: set[str] = set(known_ids or set()) - - for search_dir in search_paths: - if not os.path.isdir(search_dir): - log.debug("Module search path does not exist: %s", search_dir) - continue - - for entry in sorted(os.listdir(search_dir)): - mod_dir = os.path.join(search_dir, entry) - manifest_path = os.path.join(mod_dir, "manifest.json") - - if not os.path.isfile(manifest_path): - continue - - try: - with open(manifest_path, "r", encoding="utf-8") as f: - raw = json.load(f) - except (json.JSONDecodeError, OSError) as e: - log.warning("Skipping %s: failed to read manifest: %s", mod_dir, e) - continue - - try: - info = validate_manifest(raw, mod_dir, builtin=False) - except ManifestError: - log.warning("Skipping %s: invalid manifest", mod_dir) - continue - - if info.id in seen_ids: - log.warning( - "Skipping duplicate module '%s' at %s (already loaded from another path)", - info.id, mod_dir, - ) - continue - - info.enabled = info.id not in disabled_ids - seen_ids.add(info.id) - modules.append(info) - log.info( - "Discovered module: %s v%s (%s)%s", - info.id, info.version, "built-in" if info.builtin else "community", - "" if info.enabled else " [disabled]", - ) - return modules +log = logging.getLogger("docsis.modules") def discover_builtin_modules( - builtin_base_path: str, - disabled_ids: set[str] | None = None, + builtin_base_path: str, disabled_ids: set[str] | None = None ) -> list[ModuleInfo]: - """Load built-in module manifests from the static application registry.""" - if disabled_ids is None: - disabled_ids = set() - - modules: list[ModuleInfo] = [] - seen_ids: set[str] = set() - for entry in BUILTIN_MODULE_DIRS: - mod_dir = os.path.join(builtin_base_path, entry) - manifest_path = os.path.join(mod_dir, "manifest.json") - try: - with open(manifest_path, "r", encoding="utf-8") as f: - raw = json.load(f) - except (json.JSONDecodeError, OSError) as e: - log.warning("Skipping built-in module %s: failed to read manifest: %s", entry, e) - continue - - try: - info = validate_manifest(raw, mod_dir, builtin=True) - except ManifestError: - log.warning("Skipping built-in module %s: invalid manifest", entry) - continue - - if info.id in seen_ids: - log.warning("Skipping duplicate built-in module '%s' at %s", info.id, mod_dir) - continue - - info.enabled = info.id not in disabled_ids - seen_ids.add(info.id) - modules.append(info) - log.info( - "Registered built-in module: %s v%s%s", - info.id, - info.version, - "" if info.enabled else " [disabled]", - ) - - return modules - - -def discover_builtin_theme_modules(disabled_ids: set[str] | None = None) -> list[ModuleInfo]: - """Load application-owned themes from the static theme registry.""" - if disabled_ids is None: - disabled_ids = set() - - modules: list[ModuleInfo] = [] - seen_ids: set[str] = set() - for theme in BUILTIN_THEMES: - info = ModuleInfo( - id=theme["id"], - name=theme["name"], - description=theme["description"], - version=theme["version"], - author=theme["author"], - min_app_version=theme["minAppVersion"], - type="theme", - contributes={"theme": "builtin"}, - path="", - builtin=True, - homepage=theme.get("homepage", ""), - license=theme.get("license", ""), - menu={"order": 999}, - theme_data=theme["theme_data"], - ) - - if info.id in seen_ids: - log.warning("Skipping duplicate built-in theme '%s'", info.id) - continue - - info.enabled = info.id not in disabled_ids - seen_ids.add(info.id) - modules.append(info) - log.info( - "Registered built-in theme: %s v%s%s", - info.id, - info.version, - "" if info.enabled else " [disabled]", - ) - - return modules - - -def discover_builtin_threshold_modules(disabled_ids: set[str] | None = None) -> list[ModuleInfo]: - """Load application-owned threshold profiles from the static registry.""" - if disabled_ids is None: - disabled_ids = set() - - modules: list[ModuleInfo] = [] - seen_ids: set[str] = set() - for profile in BUILTIN_THRESHOLD_PROFILES: - info = ModuleInfo( - id=cast(str, profile["id"]), - name=cast(str, profile["name"]), - description=cast(str, profile["description"]), - version=cast(str, profile["version"]), - author=cast(str, profile["author"]), - min_app_version=cast(str, profile["minAppVersion"]), - type="analysis", - contributes={"thresholds": "builtin"}, - path="", - builtin=True, - menu={"order": 999}, - thresholds_data=deepcopy(cast(dict[str, object], profile["thresholds"])), - ) - - if info.id in seen_ids: - log.warning("Skipping duplicate built-in threshold profile '%s'", info.id) - continue - - info.enabled = info.id not in disabled_ids - seen_ids.add(info.id) - modules.append(info) - log.info( - "Registered built-in threshold profile: %s v%s%s", - info.id, - info.version, - "" if info.enabled else " [disabled]", - ) - - return modules - - -def register_module_config( - config_defaults: dict[str, Any], - module_id: str | None = None, - builtin: bool = False, - config_secrets: list[str] | None = None, - config_private: list[str] | None = None, -) -> set[str]: - """Register a module's config defaults into the global config system. - - Private-but-displayable metadata is trusted only for built-in modules. - Community secret defaults are accepted only after deterministic ownership - reservation. Core secret, private, and hash-backed settings stay - unavailable to community modules. - """ - secret_keys = set(config_secrets or []) - private_keys = set(config_private or []) if builtin else set() - registered_keys: set[str] = set() - for key, value in config_defaults.items(): - if ( - key in secret_keys - and not builtin - and _cfg.MODULE_SECRET_OWNERS.get(key) != module_id - ): - log.warning("Skipping an unreserved module secret config key") - continue - if key in _cfg.DEFAULTS: - if key in private_keys: - _cfg.PRIVATE_KEYS.add(key) - log.debug("Config key already exists in core, skipping") - continue - if not builtin and key in ( - _cfg.SECRET_KEYS | _cfg.PRIVATE_KEYS | _cfg.HASH_KEYS - ): - log.warning("Skipping a reserved core protected config key") - continue - _cfg.DEFAULTS[key] = value - registered_keys.add(key) - if key in secret_keys and builtin: - _cfg.SECRET_KEYS.add(key) - if key in private_keys: - _cfg.PRIVATE_KEYS.add(key) - if _cfg.is_secret_key(key): - # Secret values are encrypted strings at rest. Never route them - # through bool/int coercion even if a non-manifest caller supplied - # an invalid non-string default. - continue - if isinstance(value, bool): - _cfg.BOOL_KEYS.add(key) - elif isinstance(value, int): - _cfg.INT_KEYS.add(key) - return registered_keys - - -def evaluate_module_secret_ownership( - modules: list[ModuleInfo], -) -> tuple[set[str], dict[str, str], dict[str, str]]: - """Return reserved keys, valid owners, and fail-closed module errors. - - A secret key is valid only when the declaring module is also the sole - community module that declares that key in ``config``. This prevents a - disabled or newly installed module from taking over another module's plain - configuration value by reclassifying the shared key as its own secret. - """ - protected = ( - set(_cfg.CORE_CONFIG_KEYS) - | _cfg.SECRET_KEYS - | _cfg.HASH_KEYS - | _cfg.PRIVATE_KEYS - ) - for module in modules: - if module.builtin: - protected.update(module.config) - - secret_claims: dict[str, list[ModuleInfo]] = defaultdict(list) - config_claims: dict[str, list[ModuleInfo]] = defaultdict(list) - for module in modules: - if module.builtin: - continue - for key in module.config: - config_claims[key].append(module) - for key in module.config_secrets: - secret_claims[key].append(module) - - reserved_keys = {key for key in secret_claims if key not in protected} - owners: dict[str, str] = {} - errors: dict[str, str] = {} - for key, claimants in secret_claims.items(): - if key in protected: - for module in claimants: - errors[module.id] = ( - "Module secret declaration conflicts with protected configuration" - ) - continue - - config_users = config_claims.get(key, []) - valid_owner = ( - len(claimants) == 1 - and len(config_users) == 1 - and config_users[0].id == claimants[0].id - ) - if not valid_owner: - for module in [*claimants, *config_users]: - errors[module.id] = "Module secret ownership conflict" - continue - - owners[key] = claimants[0].id - - return reserved_keys, owners, errors - - -def reserve_module_secrets(modules: list[ModuleInfo]) -> None: - """Reserve all community secret declarations before any module loads.""" - reserved_keys, owners, errors = evaluate_module_secret_ownership(modules) - for module in modules: - if module.id in errors: - module.error = errors[module.id] - - protected_count = sum( - 1 for error in errors.values() if "protected configuration" in error - ) - conflict_count = len(errors) - protected_count - if protected_count: - log.warning( - "Rejected protected module secret declarations from %d module(s)", - protected_count, - ) - if conflict_count: - log.warning( - "Rejected conflicting module secret declarations affecting %d module(s)", - conflict_count, + try: + return _module_registry.discover_builtin_modules( + builtin_base_path, + disabled_ids=disabled_ids, + module_dirs=BUILTIN_MODULE_DIRS, ) + except ModuleRegistryError as exc: + raise RegistrationError(str(exc)) from exc - _cfg.set_module_secret_registry(reserved_keys, owners) - - -def merge_module_i18n(module_id: str, i18n_dir: str) -> None: - """Merge a module's i18n JSON files into the global translation system. - - Keys are namespaced under the module ID: - module i18n key "greeting" -> global key "module_id.greeting" - - Built-in modules keep ``en.json`` as their source catalog. Any existing - core language without a module-specific catalog receives the English module - strings as its fallback so templates can keep using one translation dict per - request. If a community module ships its own locale file, those strings - overlay the English fallback for that language. - """ - if not os.path.isdir(i18n_dir): - log.debug("No i18n directory for module '%s': %s", module_id, i18n_dir) - return - - catalogs = {} - for fname in sorted(os.listdir(i18n_dir)): - if not fname.endswith(".json") or fname == "template.json": - continue - lang = fname[:-5] # "en.json" -> "en" - fpath = os.path.join(i18n_dir, fname) - try: - with open(fpath, "r", encoding="utf-8") as f: - data = json.load(f) - except (json.JSONDecodeError, OSError) as e: - log.warning("Failed to load i18n file %s: %s", fpath, e) - continue - - if not isinstance(data, dict): - log.warning("Skipping non-object i18n payload in %s", fpath) - continue - catalogs[lang] = data - - if not catalogs: - return - - fallback = catalogs.get("en", {}) - target_langs = set(_TRANSLATIONS) | set(catalogs) - if fallback: - target_langs.add("en") - - for lang in sorted(target_langs): - data = dict(fallback) - if lang != "en": - data.update(catalogs.get(lang, {})) - elif "en" in catalogs: - data = catalogs["en"] - elif not data: - continue - - if lang not in _TRANSLATIONS: - _TRANSLATIONS[lang] = {} - - merged = 0 - for key, value in data.items(): - if key.startswith("_"): - continue # skip metadata keys like _meta - _TRANSLATIONS[lang][f"{module_id}.{key}"] = value - merged += 1 - # Also add un-namespaced key for backward compat with JS code. - if key not in _TRANSLATIONS[lang]: - _TRANSLATIONS[lang][key] = value - log.debug("Merged %d i18n keys for module '%s' lang '%s'", merged, module_id, lang) - - -# Core routes that community modules must not shadow. These are exact -# paths (or prefixes ending with /) that protect authentication, config, -# and core data endpoints from being intercepted by untrusted code. -_PROTECTED_ROUTES = { - "/", "/login", "/logout", "/setup", "/settings", "/health", "/sw.js", -} -_PROTECTED_API_PREFIXES = ( - "/api/config", "/api/data", "/api/tokens", "/api/demo", - "/api/poll", "/api/status", "/api/history", "/api/events", "/api/trends", - "/api/export", "/api/correlation", - "/api/modules/", "/api/themes/", -) - - -def _load_symbol(spec: str, module_id: str): - """Import a trusted built-in Python contribution by module path.""" - if ":" not in spec: - log.warning("Built-in module '%s': invalid Python contribution spec", module_id) - return None - module_name, attr_name = spec.rsplit(":", 1) +def discover_builtin_theme_modules( + disabled_ids: set[str] | None = None, +) -> list[ModuleInfo]: try: - module = importlib.import_module(module_name) - except Exception as e: - log.error("Built-in module '%s': failed to import %s: %s", module_id, module_name, e) - return None - value = getattr(module, attr_name, None) - if value is None: - log.warning("Built-in module '%s': symbol '%s' not found in %s", module_id, attr_name, module_name) - return value - - -def attach_builtin_python_contributions(mod: ModuleInfo) -> None: - """Attach statically registered Python entry points for a built-in module.""" - specs = BUILTIN_PYTHON_CONTRIBUTIONS.get(mod.id) - if not specs: - specs = None - - for key, attr_name in ( - ("collector", "collector_class"), - ("publisher", "publisher_class"), - ): - if key not in mod.contributes or getattr(mod, attr_name) is not None: - continue - spec = getattr(specs, key, None) if specs else None - if not spec: - raise ManifestError(f"Built-in module '{mod.id}' missing static {key} registration") - symbol = _load_symbol(spec, mod.id) - if symbol is None: - raise ManifestError(f"Built-in module '{mod.id}' failed to import static {key} registration") - setattr(mod, attr_name, symbol) - - -def load_module_routes(app, module_id: str, module_path: str, routes_file: str, *, builtin: bool = False) -> None: - """Dynamically load a Flask Blueprint from a module's routes file. - - The routes file must export a variable named 'bp' or 'blueprint' - that is a Flask Blueprint instance. - - Community modules (builtin=False) are checked for route conflicts - with core endpoints before registration. - """ - routes_path = safe_manifest_ref(module_path, routes_file) - if not os.path.isfile(routes_path): - log.warning("Module '%s': routes file not found: %s", module_id, routes_path) - return - - dir_name = os.path.basename(module_path) - if builtin: - mod_name = f"app.modules.{dir_name}.{os.path.splitext(os.path.basename(routes_file))[0]}" - try: - mod = importlib.import_module(mod_name) - except Exception as e: - log.error("Module '%s': failed to import routes: %s", module_id, e) - return - else: - mod_name = f"community_modules.{dir_name}.routes" - try: - spec = importlib.util.spec_from_file_location(mod_name, routes_path) - if spec is None or spec.loader is None: - log.warning("Module '%s': could not create import spec for %s", module_id, routes_path) - return - mod = importlib.util.module_from_spec(spec) - sys.modules[mod_name] = mod - spec.loader.exec_module(mod) - except Exception as e: - log.error("Module '%s': failed to import routes: %s", module_id, e) - return - - # Find Blueprint - blueprint = getattr(mod, "bp", None) or getattr(mod, "blueprint", None) - if blueprint is None: - log.warning("Module '%s': routes.py does not export 'bp' or 'blueprint'", module_id) - return - - # Community module route validation: block blueprints that shadow - # core routes (prevents login interception, config hijacking, etc.). - if not builtin: - rules_before = set(r.rule for r in app.url_map.iter_rules()) - try: - app.register_blueprint(blueprint) - except Exception as e: - log.error("Module '%s': failed to register blueprint: %s", module_id, e) - return - rules_after = set(r.rule for r in app.url_map.iter_rules()) - new_rules = rules_after - rules_before - blocked = [] - for rule in new_rules: - if rule in _PROTECTED_ROUTES: - blocked.append(rule) - continue - for prefix in _PROTECTED_API_PREFIXES: - if rule.startswith(prefix): - blocked.append(rule) - break - if blocked: - log.error( - "Module '%s': BLOCKED -- routes conflict with core endpoints: %s. " - "Community modules must not shadow protected routes.", - module_id, ", ".join(sorted(blocked)), - ) - return - log.info("Module '%s': registered community routes blueprint (%d routes)", module_id, len(new_rules)) - else: - try: - app.register_blueprint(blueprint) - log.info("Module '%s': registered routes blueprint", module_id) - except Exception as e: - log.error("Module '%s': failed to register blueprint: %s", module_id, e) - - -def _load_module_class(module_id: str, module_path: str, spec: str, kind: str): - """Load a contributed class from a module-owned Python file.""" - if ":" not in spec: - log.warning("Module '%s': %s spec must be 'file.py:ClassName', got '%s'", module_id, kind, spec) - return None - - filename, class_name = spec.rsplit(":", 1) - file_path = safe_manifest_ref(module_path, filename) + return _module_registry.discover_builtin_theme_modules( + disabled_ids=disabled_ids, + themes=BUILTIN_THEMES, + ) + except ModuleRegistryError as exc: + raise RegistrationError(str(exc)) from exc - if not os.path.isfile(file_path): - log.warning("Module '%s': %s file not found: %s", module_id, kind, file_path) - return None - dir_name = os.path.basename(module_path) - mod_name = f"app.modules.{dir_name}.{kind}" +def discover_builtin_threshold_modules( + disabled_ids: set[str] | None = None, +) -> list[ModuleInfo]: try: - im_spec = importlib.util.spec_from_file_location(mod_name, file_path) - if im_spec is None or im_spec.loader is None: - log.warning("Module '%s': could not create import spec for %s", module_id, file_path) - return None - mod = importlib.util.module_from_spec(im_spec) - sys.modules[mod_name] = mod - im_spec.loader.exec_module(mod) - except Exception as e: - log.error("Module '%s': failed to import %s: %s", module_id, kind, e) - return None - - cls = getattr(mod, class_name, None) - if cls is None: - log.warning("Module '%s': class '%s' not found in %s", module_id, class_name, file_path) - return None - - log.info("Module '%s': loaded %s class '%s'", module_id, kind, class_name) - return cls - - -def load_module_collector(module_id: str, module_path: str, spec: str): - """Load a Collector class from a module file. - - Args: - module_id: The module's unique identifier. - module_path: Filesystem path to the module directory. - spec: "filename.py:ClassName" format (e.g. "collector.py:WeatherCollector") - - Returns: - The Collector subclass, or None if loading failed. - """ - return _load_module_class(module_id, module_path, spec, "collector") - - -def load_module_publisher(module_id: str, module_path: str, spec: str): - """Load a Publisher class from a module file. - - Args: - module_id: The module's unique identifier. - module_path: Filesystem path to the module directory. - spec: "filename.py:ClassName" format (e.g. "publisher.py:MQTTPublisher") - - Returns: - The Publisher class, or None if loading failed. - """ - return _load_module_class(module_id, module_path, spec, "publisher") - - -def module_static_endpoint(module_id: str) -> str: - """Return the stable Flask endpoint name for a module's static files.""" - return f"module_static_{module_id}" - - -def module_static_url(module_id: str, filename: str, **values: Any) -> str: - """Build a module-static URL using the registered endpoint contract.""" - return url_for(module_static_endpoint(module_id), filename=filename, **values) - - -def setup_module_static(app, module_id: str, module_path: str, static_subdir: str) -> None: - """Mount a module's static directory at /modules//static/.""" - static_dir = safe_manifest_subpath(module_path, static_subdir.rstrip("/")) - if not os.path.isdir(static_dir): - log.debug("Module '%s': no static directory at %s", module_id, static_dir) - return - - static_root = os.path.realpath(static_dir) - static_prefix = static_root + os.sep - route = f"/modules/{module_id}/static/" - - def serve_static(filename, _root=static_root, _prefix=static_prefix): - try: - candidate = os.path.realpath(os.path.join(_root, filename)) - except (OSError, TypeError, ValueError): - return abort(404) - if candidate.startswith(_prefix): - if not os.path.isfile(candidate): - return abort(404) - return send_file(candidate) - return abort(404) - - # Use a unique endpoint name per module - endpoint = module_static_endpoint(module_id) - app.add_url_rule(route, endpoint=endpoint, view_func=serve_static) - log.info("Module '%s': serving static files at /modules/%s/static/", module_id, module_id) - - -def setup_module_templates( - module_id: str, module_path: str, contributes: dict[str, str] -) -> dict[str, str]: - """Resolve module template paths to absolute file paths. - - Args: - contributes: Dict with keys like 'tab', 'card', 'settings' mapping to - relative template paths within the module directory. - - Returns: - Dict of template type -> absolute file path (only for files that exist). - """ - template_keys = {"tab", "card", "settings"} - resolved = {} - - for key in template_keys: - rel_path = contributes.get(key) - if not rel_path: - continue - abs_path = safe_manifest_subpath(module_path, rel_path) - if os.path.isfile(abs_path): - # Store just the filename for Jinja2 include (ChoiceLoader resolves it) - resolved[key] = os.path.basename(abs_path) - log.debug("Module '%s': template '%s' -> %s", module_id, key, abs_path) - else: - log.warning("Module '%s': template '%s' not found: %s", module_id, key, abs_path) - - return resolved - - -REQUIRED_THRESHOLD_SECTIONS = {"downstream_power", "upstream_power", "snr"} - - -def validate_thresholds(data: dict[str, object]) -> None: - """Validate a threshold JSON structure. - - Raises ManifestError if required sections or keys are missing. - """ - missing = REQUIRED_THRESHOLD_SECTIONS - set(data.keys()) - if missing: - raise ManifestError(f"Missing required threshold sections: {', '.join(sorted(missing))}") - - for section in REQUIRED_THRESHOLD_SECTIONS: - block = data[section] - if not isinstance(block, dict): - raise ManifestError(f"Threshold section '{section}' must be a dict") - if "_default" not in block: - raise ManifestError(f"Threshold section '{section}' missing '_default' key") - - -REQUIRED_THEME_SECTIONS = {"dark", "light"} - - -def validate_theme(data: dict[str, object]) -> None: - """Validate a theme.json structure. - - Raises ManifestError if required sections are missing or values are invalid. - """ - missing = REQUIRED_THEME_SECTIONS - set(data.keys()) - if missing: - raise ManifestError(f"Missing required theme sections: {', '.join(sorted(missing))}") - - for section in REQUIRED_THEME_SECTIONS: - block = data[section] - if not isinstance(block, dict): - raise ManifestError(f"Theme section '{section}' must be a dict") - if not block: - raise ManifestError(f"Theme section '{section}' is empty") - for key, value in block.items(): - if not isinstance(value, str): - raise ManifestError( - f"Theme property '{key}' in '{section}' must be a string, got {type(value).__name__}" - ) + return _module_registry.discover_builtin_threshold_modules( + disabled_ids=disabled_ids, + profiles=BUILTIN_THRESHOLD_PROFILES, + ) + except ModuleRegistryError as exc: + raise RegistrationError(str(exc)) from exc class ModuleLoader: - """Orchestrates module discovery, validation, and loading. - - Usage: - loader = ModuleLoader(app, search_paths=[...]) - modules = loader.load_all() - """ + """Discover modules and assemble their complete validated registration plan.""" - def __init__( - self, - app, - search_paths: list[str] | None = None, - disabled_ids: set[str] | None = None, - builtin_base_path: str | None = None, - ): + def __init__(self, app, search_paths=None, disabled_ids=None, builtin_base_path=None): self._app = app self._search_paths = search_paths or [] self._disabled_ids = disabled_ids or set() self._builtin_base_path = builtin_base_path self._modules: list[ModuleInfo] = [] + self._registration_plan = RegistrationPlan() def load_all(self) -> list[ModuleInfo]: - """Discover and load all modules. - - Returns list of all discovered ModuleInfo (including disabled). - """ + """Discover, resolve, and deterministically accept complete module plans.""" modules: list[ModuleInfo] = [] if self._builtin_base_path: - modules.extend( - discover_builtin_modules( - self._builtin_base_path, - disabled_ids=self._disabled_ids, - ) - ) - modules.extend( - discover_builtin_threshold_modules(disabled_ids=self._disabled_ids) - ) - modules.extend( - discover_builtin_theme_modules(disabled_ids=self._disabled_ids) - ) - modules.extend( - discover_modules( - search_paths=self._search_paths, - disabled_ids=self._disabled_ids, - known_ids={m.id for m in modules}, - ) - ) + modules.extend(discover_builtin_modules( + self._builtin_base_path, disabled_ids=self._disabled_ids, + )) + modules.extend(discover_builtin_threshold_modules(self._disabled_ids)) + modules.extend(discover_builtin_theme_modules(self._disabled_ids)) + modules.extend(discover_modules( + search_paths=self._search_paths, disabled_ids=self._disabled_ids, + known_ids={mod.id for mod in modules}, + )) self._modules = modules - - # Built-in private metadata is a separate, displayable classification. - # Register it before community secret claims are evaluated. - for mod in self._modules: - if mod.builtin and mod.config_private: - _cfg.PRIVATE_KEYS.update(mod.config_private) - reserve_module_secrets(self._modules) - - for mod in self._modules: - # Privacy classification must survive module disablement so values - # already stored by a built-in module remain encrypted/decryptable. + builtin_ids = [mod.id for mod in modules if mod.builtin] + if len(builtin_ids) != len(set(builtin_ids)): + raise RegistrationError("Duplicate built-in module id") + _reserved, _owners, ownership_errors = evaluate_module_secret_ownership(modules) + config_errors = evaluate_module_config_ownership(modules) + for mod in modules: + mod.error = ownership_errors.get(mod.id) or config_errors.get(mod.id) + resolved = [] + for mod in modules: if mod.error: - log.warning("Module '%s' rejected before load", mod.id) + if mod.builtin: + raise RegistrationError( + f"Built-in module {mod.id} failed ownership validation" + ) continue if not mod.enabled: - # Theme modules: load theme_data even when disabled so - # the settings gallery can show previews for all themes. - if mod.type == "theme" and "theme" in mod.contributes: + if not mod.builtin and mod.type == "theme" and "theme" in mod.contributes: try: - theme_path = safe_manifest_ref( - mod.path, mod.contributes["theme"] + mod.theme_data = _read_json_contribution( + mod.path, mod.contributes["theme"], "theme", validate_theme ) - if os.path.isfile(theme_path): - with open(theme_path, "r", encoding="utf-8") as f: - tdata = json.load(f) - validate_theme(tdata) - mod.theme_data = tdata - except Exception as e: - log.warning( - "Module '%s': theme preview load failed: %s", - mod.id, e, + except Exception as exc: + mod.error = ( + str(exc) if isinstance(exc, ManifestError) + else "theme contribution is invalid" ) + log.warning("Module '%s': disabled theme preview rejected", mod.id) log.info("Module '%s' is disabled, skipping load", mod.id) continue - try: - self._load_module(mod) - except Exception as e: - mod.error = str(e) - log.error("Module '%s' failed to load: %s", mod.id, e) - - enabled = [m for m in self._modules if m.enabled and not m.error] + resolved.append(resolve_module_contribution(mod)) + except Exception as exc: + if mod.builtin: + raise RegistrationError( + f"Built-in module {mod.id} failed contribution preflight: " + f"{type(exc).__name__}" + ) from exc + mod.error = str(exc) + log.error("Module '%s' failed contribution preflight", mod.id) + accepted = self._validate_resolved(resolved) + accepted_ids = {item.module_id for item, _http in accepted} + registry_modules = [ + mod for mod in modules + if mod.id in accepted_ids or not mod.enabled or mod.id in ownership_errors + ] + secret_keys, secret_owners, _errors = evaluate_module_secret_ownership( + registry_modules + ) + http = RegistrationPlan().combined(*(plan for _item, plan in accepted)) + self._registration_plan = RegistrationPlan( + rules=http.rules, blueprints=http.blueprints, + modules=tuple(item for item, _http in accepted), + module_secret_keys=tuple(sorted(secret_keys)), + module_secret_owners=tuple(sorted(secret_owners.items())), + builtin_private_keys=tuple(sorted( + key for mod in modules if mod.builtin for key in mod.config_private + )), + ) + plan = self._registration_plan + if ( + any((plan.rules, plan.blueprints, plan.modules, plan.module_secret_keys, + plan.builtin_private_keys)) + and not self._app.extensions.get("docsight_registration_deferred", False) + ): + register_plan(self._app, plan) + enabled = [mod for mod in self._modules if mod.enabled and not mod.error] log.info( "Module loading complete: %d discovered, %d enabled, %d failed", - len(self._modules), - len(enabled), - len([m for m in self._modules if m.error]), + len(self._modules), len(enabled), + len([mod for mod in self._modules if mod.error]), ) - return self._modules - def _load_module(self, mod: ModuleInfo) -> None: - """Load a single module's contributions.""" - c = mod.contributes - - if mod.builtin: - attach_builtin_python_contributions(mod) - - # Config defaults - if mod.config: - register_module_config( - mod.config, - module_id=mod.id, - builtin=mod.builtin, - config_secrets=mod.config_secrets, - config_private=mod.config_private, - ) - - # i18n - if "i18n" in c: - i18n_dir = safe_manifest_subpath(mod.path, c["i18n"].rstrip("/")) - merge_module_i18n(mod.id, i18n_dir) - - # Routes (Blueprint) - if "routes" in c: - load_module_routes(self._app, mod.id, mod.path, c["routes"], builtin=mod.builtin) - - # Static files and convention-based asset detection - static_subdir = c.get("static", "static/").rstrip("/") - static_dir = safe_manifest_subpath(mod.path, static_subdir) - if os.path.isdir(static_dir): - setup_module_static(self._app, mod.id, mod.path, static_subdir) - mod.has_css = os.path.isfile(os.path.join(static_dir, "style.css")) - mod.has_js = os.path.isfile(os.path.join(static_dir, "main.js")) - - # Template paths - mod.template_paths = setup_module_templates(mod.id, mod.path, c) - - # Collector (class loaded but not instantiated -- collector discovery handles that) - if "collector" in c and not mod.builtin: - mod.collector_class = load_module_collector(mod.id, mod.path, c["collector"]) - - # Publisher (class loaded but not instantiated -- main.py handles that) - if "publisher" in c and not mod.builtin: - mod.publisher_class = load_module_publisher(mod.id, mod.path, c["publisher"]) - - # Thresholds - if "thresholds" in c: - # Built-in profiles preload thresholds_data and use the "builtin" sentinel; - # community profiles keep the manifest-relative thresholds file path. - if mod.thresholds_data is None: - thresholds_path = safe_manifest_ref(mod.path, c["thresholds"]) - if not os.path.isfile(thresholds_path): - raise ManifestError(f"Thresholds file not found: {c['thresholds']}") - with open(thresholds_path, "r", encoding="utf-8") as f: - tdata = json.load(f) - validate_thresholds(tdata) - mod.thresholds_data = tdata - else: - tdata = mod.thresholds_data - validate_thresholds(tdata) - _analyzer.set_thresholds( - tdata, - profile_id=mod.id, - profile_version=mod.version, + def _validate_resolved(self, resolved): + base = self._app.extensions.get( + "docsight_registration_base_plan", RegistrationPlan() + ) + accepted = [] + for item, http in resolved: + candidate = base.combined( + *(current_http for _current, current_http in accepted), http, + RegistrationPlan(modules=tuple( + current for current, _current_http in accepted + ) + (item,)), ) - log.info("Module '%s': loaded threshold profile", mod.id) + try: + validate_plan( + candidate, + existing=existing_rules(self._app), + existing_blueprints=tuple(self._app.blueprints), + ) + except RegistrationError as exc: + if item.builtin: + raise RegistrationError( + f"Built-in module {item.module_id} has a registration collision" + ) from exc + item.info.error = str(exc) + continue + accepted.append((item, http)) + return accepted - # Theme - if "theme" in c: - if mod.theme_data is None: - theme_path = safe_manifest_ref(mod.path, c["theme"]) - if not os.path.isfile(theme_path): - raise ManifestError(f"Theme file not found: {c['theme']}") - with open(theme_path, "r", encoding="utf-8") as f: - tdata = json.load(f) - validate_theme(tdata) - mod.theme_data = tdata - else: - validate_theme(mod.theme_data) - log.info("Module '%s': loaded theme profile", mod.id) + @property + def registration_plan(self) -> RegistrationPlan: + return self._registration_plan def get_modules(self) -> list[ModuleInfo]: - """Return all discovered modules (enabled and disabled).""" return list(self._modules) def get_enabled_modules(self) -> list[ModuleInfo]: - """Return only enabled modules without errors.""" - return [m for m in self._modules if m.enabled and not m.error] + return [mod for mod in self._modules if mod.enabled and not mod.error] def get_threshold_modules(self) -> list[ModuleInfo]: - """Return all modules that contribute thresholds.""" - return [m for m in self._modules if "thresholds" in m.contributes] + return [mod for mod in self._modules if "thresholds" in mod.contributes] def get_theme_modules(self) -> list[ModuleInfo]: - """Return all modules that contribute theme definitions.""" - return [m for m in self._modules if "theme" in m.contributes] + return [mod for mod in self._modules if "theme" in mod.contributes] diff --git a/app/module_registry.py b/app/module_registry.py new file mode 100644 index 00000000..864b59a5 --- /dev/null +++ b/app/module_registry.py @@ -0,0 +1,194 @@ +"""Module manifest models and deterministic source discovery.""" + +from __future__ import annotations + +import json +import logging +import os +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any, cast + +from .builtin_modules import BUILTIN_MODULE_DIRS +from .manifest_contract import validate_manifest_contract +from .theme_registry import BUILTIN_THEMES +from .threshold_profiles import BUILTIN_THRESHOLD_PROFILES + + +log = logging.getLogger("docsis.modules") + + +class ManifestError(Exception): + """Raised when a module manifest is invalid.""" + + +class ModuleRegistryError(RuntimeError): + """Raised when the application-owned module registry is inconsistent.""" + + +@dataclass +class ModuleInfo: + """Validated module metadata from manifest.json.""" + + id: str + name: str + description: str + version: str + author: str + min_app_version: str + type: str + contributes: dict[str, str] + path: str + builtin: bool = False + homepage: str = "" + license: str = "" + config: dict[str, Any] = field(default_factory=dict) + config_secrets: list[str] = field(default_factory=list) + config_private: list[str] = field(default_factory=list) + menu: dict[str, Any] = field(default_factory=dict) + enabled: bool = True + error: str | None = None + template_paths: dict[str, str] = field(default_factory=dict) + collector_class: type | None = None + publisher_class: type | None = None + hints: dict[str, object] = field(default_factory=dict) + thresholds_data: dict[str, object] | None = None + theme_data: dict[str, object] | None = None + has_css: bool = False + has_js: bool = False + + +def validate_manifest( + raw: dict[str, Any], module_path: str, *, builtin: bool | None = None +) -> ModuleInfo: + """Validate a raw manifest dict and return stable module metadata.""" + if builtin is None: + norm = os.path.normpath(module_path).replace("\\", "/") + builtin = "/app/modules/" in norm + errors = validate_manifest_contract(raw, builtin=builtin) + if errors: + raise ManifestError(errors[0]) + return ModuleInfo( + id=raw["id"], + name=raw["name"], + description=raw["description"], + version=raw["version"], + author=raw["author"], + min_app_version=raw["minAppVersion"], + type=raw["type"], + contributes=raw["contributes"], + path=module_path, + builtin=builtin, + homepage=raw.get("homepage", ""), + license=raw.get("license", ""), + config=raw.get("config", {}), + config_secrets=raw.get("config_secrets", []), + config_private=raw.get("configPrivate", []), + menu={**{"order": 999}, **raw.get("menu", {})}, + hints=raw.get("hints", {}), + ) + + +def _manifest(path: str, *, builtin: bool) -> ModuleInfo: + try: + with open(os.path.join(path, "manifest.json"), "r", encoding="utf-8") as handle: + raw = json.load(handle) + except (json.JSONDecodeError, OSError) as exc: + if builtin: + raise ModuleRegistryError("Built-in module manifest could not be read") from exc + raise ManifestError("Module manifest could not be read") from exc + return validate_manifest(raw, path, builtin=builtin) + + +def discover_modules( + search_paths: list[str] | None = None, + disabled_ids: set[str] | None = None, + known_ids: set[str] | None = None, +) -> list[ModuleInfo]: + """Discover community modules in stable path and directory order.""" + modules: list[ModuleInfo] = [] + disabled = disabled_ids or set() + seen = set(known_ids or set()) + for search_dir in search_paths or []: + if not os.path.isdir(search_dir): + continue + for entry in sorted(os.listdir(search_dir)): + mod_dir = os.path.join(search_dir, entry) + if not os.path.isfile(os.path.join(mod_dir, "manifest.json")): + continue + try: + info = _manifest(mod_dir, builtin=False) + except ManifestError: + log.warning("Skipping invalid community module manifest") + continue + if info.id in seen: + log.warning("Skipping duplicate module '%s'; first source wins", info.id) + continue + info.enabled = info.id not in disabled + seen.add(info.id) + modules.append(info) + return modules + + +def discover_builtin_modules( + builtin_base_path: str, + disabled_ids: set[str] | None = None, + module_dirs: tuple[str, ...] | None = None, +) -> list[ModuleInfo]: + """Resolve the statically ordered built-in manifest registry.""" + modules: list[ModuleInfo] = [] + for entry in module_dirs or BUILTIN_MODULE_DIRS: + try: + modules.append(_manifest(os.path.join(builtin_base_path, entry), builtin=True)) + except ManifestError as exc: + raise ModuleRegistryError(f"Built-in module {entry} manifest is invalid") from exc + return _finalize_builtins(modules, disabled_ids, "module") + + +def _finalize_builtins( + modules: list[ModuleInfo], disabled_ids: set[str] | None, kind: str +) -> list[ModuleInfo]: + seen, disabled = set(), disabled_ids or set() + for info in modules: + if info.id in seen: + raise ModuleRegistryError(f"Duplicate built-in {kind} id {info.id}") + info.enabled = info.id not in disabled + seen.add(info.id) + return modules + + +def discover_builtin_theme_modules( + disabled_ids: set[str] | None = None, + themes=None, +) -> list[ModuleInfo]: + """Resolve application-owned themes from their static registry.""" + modules: list[ModuleInfo] = [] + for theme in BUILTIN_THEMES if themes is None else themes: + modules.append(ModuleInfo( + id=theme["id"], name=theme["name"], description=theme["description"], + version=theme["version"], author=theme["author"], + min_app_version=theme["minAppVersion"], type="theme", + contributes={"theme": "builtin"}, path="", builtin=True, + homepage=theme.get("homepage", ""), license=theme.get("license", ""), + menu={"order": 999}, theme_data=theme["theme_data"], + )) + return _finalize_builtins(modules, disabled_ids, "theme") + + +def discover_builtin_threshold_modules( + disabled_ids: set[str] | None = None, + profiles=None, +) -> list[ModuleInfo]: + """Resolve application-owned threshold profiles from their static registry.""" + modules: list[ModuleInfo] = [] + for profile in BUILTIN_THRESHOLD_PROFILES if profiles is None else profiles: + modules.append(ModuleInfo( + id=cast(str, profile["id"]), name=cast(str, profile["name"]), + description=cast(str, profile["description"]), + version=cast(str, profile["version"]), author=cast(str, profile["author"]), + min_app_version=cast(str, profile["minAppVersion"]), type="analysis", + contributes={"thresholds": "builtin"}, path="", builtin=True, + menu={"order": 999}, + thresholds_data=deepcopy(cast(dict[str, object], profile["thresholds"])), + )) + return _finalize_builtins(modules, disabled_ids, "threshold profile") diff --git a/app/registration.py b/app/registration.py new file mode 100644 index 00000000..6105c508 --- /dev/null +++ b/app/registration.py @@ -0,0 +1,350 @@ +"""Immutable planning and atomic application of DOCSight registrations.""" + +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass, field +from typing import Any, Callable, Iterable, Sequence + +from flask import Blueprint, Flask +from jinja2 import ChoiceLoader, FileSystemLoader + +from .i18n import _TRANSLATIONS + + +log = logging.getLogger("docsis.modules") + + +class RegistrationError(RuntimeError): + """A registration plan is invalid and must not be applied.""" + + +@dataclass(frozen=True) +class PlannedRule: + rule: str + endpoint: str + methods: tuple[str, ...] + source: str + view: Callable[..., Any] | None = field(default=None, compare=False, repr=False) + + def __post_init__(self) -> None: + methods = {method.upper() for method in self.methods} + if "GET" in methods: + methods.add("HEAD") + object.__setattr__(self, "methods", tuple(sorted(methods | {"OPTIONS"}))) + + +@dataclass(frozen=True) +class PlannedBlueprint: + name: str + source: str + blueprint: Blueprint = field(compare=False, repr=False) + rules: tuple[PlannedRule, ...] = () + registered_names: tuple[str, ...] = () + + +@dataclass(frozen=True) +class ModuleContribution: + """One module's complete, resolved contribution set.""" + + module_id: str + source: str + version: str + builtin: bool + info: Any = field(compare=False, repr=False) + config: tuple[tuple[str, Any], ...] = field(default=(), repr=False) + secret_keys: tuple[str, ...] = field(default=(), repr=False) + private_keys: tuple[str, ...] = field(default=(), repr=False) + i18n_catalogs: tuple[tuple[str, tuple[tuple[str, Any], ...]], ...] = () + template_paths: tuple[tuple[str, str], ...] = () + template_dir: str | None = field(default=None, compare=False, repr=False) + collector_class: type | None = field(default=None, compare=False, repr=False) + publisher_class: type | None = field(default=None, compare=False, repr=False) + thresholds_data: dict[str, object] | None = field(default=None, compare=False, repr=False) + theme_data: dict[str, object] | None = field(default=None, compare=False, repr=False) + has_css: bool = False + has_js: bool = False + + +@dataclass(frozen=True) +class RegistrationPlan: + """The ordered, immutable complete registration contract.""" + + rules: tuple[PlannedRule, ...] = () + blueprints: tuple[PlannedBlueprint, ...] = () + modules: tuple[ModuleContribution, ...] = () + module_secret_keys: tuple[str, ...] = field(default=(), repr=False) + module_secret_owners: tuple[tuple[str, str], ...] = field(default=(), repr=False) + builtin_private_keys: tuple[str, ...] = field(default=(), repr=False) + + def combined(self, *others: "RegistrationPlan") -> "RegistrationPlan": + plans = (self, *others) + return RegistrationPlan( + rules=tuple(item for plan in plans for item in plan.rules), + blueprints=tuple(item for plan in plans for item in plan.blueprints), + modules=tuple(item for plan in plans for item in plan.modules), + module_secret_keys=tuple(sorted({item for plan in plans for item in plan.module_secret_keys})), + module_secret_owners=tuple(sorted(dict( + item for plan in plans for item in plan.module_secret_owners + ).items())), + builtin_private_keys=tuple(sorted({ + item for plan in plans for item in plan.builtin_private_keys + })), + ) + + +def probe_blueprint(blueprint: Blueprint, *, source: str) -> PlannedBlueprint: + """Preflight a blueprint on an isolated app.""" + tree, pending = [], [blueprint] + while pending: + item = pending.pop() + tree.append(item) + pending.extend(child for child, _options in getattr(item, "_blueprints", ())) + previous = { + item: (getattr(item, "_got_registered_once", False), item.cli.name) + for item in tree + } + probe = Flask("registration.probe", static_folder=None) + try: + probe.register_blueprint(blueprint) + rules = tuple( + PlannedRule( + rule.rule, rule.endpoint, tuple(rule.methods), source, + probe.view_functions.get(rule.endpoint), + ) + for rule in probe.url_map.iter_rules() + ) + names = tuple(probe.blueprints) + except Exception as exc: + raise RegistrationError( + f"Blueprint '{blueprint.name}' from {source} failed preflight: {type(exc).__name__}" + ) from exc + finally: + for item, (was_registered, cli_name) in previous.items(): + item._got_registered_once = was_registered + item.cli.name = cli_name + return PlannedBlueprint(blueprint.name, source, blueprint, rules, names) + + +def existing_rules(app: Flask) -> tuple[PlannedRule, ...]: + return tuple( + PlannedRule( + rule.rule, rule.endpoint, tuple(rule.methods), "existing", + app.view_functions.get(rule.endpoint), + ) + for rule in app.url_map.iter_rules() + ) + + +def validate_plan( + plan: RegistrationPlan, + *, + existing: Sequence[PlannedRule] = (), + existing_blueprints: Sequence[str] = (), +) -> None: + """Reject all identity, ownership, endpoint, blueprint and route collisions.""" + errors: list[str] = [] + ids: set[str] = set() + for module in plan.modules: + if module.module_id in ids: + errors.append(f"duplicate module id '{module.module_id}'") + ids.add(module.module_id) + claims = ( + ("config key", ((key, module.module_id) for module in plan.modules for key, _ in module.config)), + ("secret ownership", ( + *((key, module.module_id) for module in plan.modules for key in module.secret_keys), + *plan.module_secret_owners, + )), + ) + for kind, values in claims: + owners: dict[str, str] = {} + for key, owner in values: + if key in owners and owners[key] != owner: + errors.append(f"{kind} collision between {owners[key]} and {owner}") + owners.setdefault(key, owner) + names = {name: "existing" for name in existing_blueprints} + for blueprint in plan.blueprints: + for name in blueprint.registered_names or (blueprint.name,): + if name in names: + errors.append( + f"blueprint name collision '{name}' between {names[name]} and {blueprint.source}" + ) + names.setdefault(name, blueprint.source) + rules = (*existing, *plan.rules, *( + rule for blueprint in plan.blueprints for rule in blueprint.rules + )) + endpoints: dict[str, PlannedRule] = {} + routes: dict[tuple[str, str], PlannedRule] = {} + for rule in rules: + owner = endpoints.get(rule.endpoint) + if owner and not (owner.source == rule.source and owner.view is rule.view): + errors.append( + f"endpoint collision '{rule.endpoint}' between {owner.source} and {rule.source}" + ) + endpoints.setdefault(rule.endpoint, rule) + for method in (method for method in rule.methods if method != "OPTIONS"): + key, owner = (rule.rule, method), routes.get((rule.rule, method)) + same = owner and owner.source == rule.source and owner.endpoint == rule.endpoint and owner.view is rule.view + if owner and not same: + errors.append( + f"route/method collision '{rule.rule}' {method} between {owner.source} and {rule.source}" + ) + routes.setdefault(key, rule) + if errors: + raise RegistrationError("; ".join(sorted(set(errors)))) + + +def apply_module_i18n(module_id: str, catalogs: dict[str, dict[str, Any]]) -> None: + """Merge validated, namespaced catalogs with English fallback.""" + if not catalogs: + return + fallback = catalogs.get("en", {}) + target_langs = set(_TRANSLATIONS) | set(catalogs) + if fallback: + target_langs.add("en") + for lang in sorted(target_langs): + data = dict(fallback) + if lang != "en": + data.update(catalogs.get(lang, {})) + elif "en" in catalogs: + data = catalogs["en"] + elif not data: + continue + _TRANSLATIONS.setdefault(lang, {}) + merged = 0 + for key, value in data.items(): + if key.startswith("_"): + continue + _TRANSLATIONS[lang][f"{module_id}.{key}"] = value + merged += 1 + _TRANSLATIONS[lang].setdefault(key, value) + log.debug( + "Merged %d i18n keys for module '%s' lang '%s'", + merged, module_id, lang, + ) + + +def register_plan(app: Flask, plan: RegistrationPlan) -> None: + """Validate and atomically apply a complete plan exactly once.""" + from . import analyzer, config as cfg + from .module_config_registry import register_module_config + + applied = app.extensions.get("docsight_applied_registration_plans", ()) + if any(plan is previous_plan for previous_plan in applied): + raise RegistrationError("Registration plan has already been applied") + previous = app.extensions.get("docsight_registration_plan", RegistrationPlan()) + validate_plan(RegistrationPlan( + modules=previous.modules + plan.modules, + module_secret_owners=previous.module_secret_owners + plan.module_secret_owners, + )) + validate_plan( + plan, existing=existing_rules(app), existing_blueprints=tuple(app.blueprints) + ) + for rule in plan.rules: + app.add_url_rule( + rule.rule, endpoint=rule.endpoint, view_func=rule.view, + methods=[method for method in rule.methods if method != "OPTIONS"], + ) + for blueprint in plan.blueprints: + app.register_blueprint(blueprint.blueprint) + + complete = previous.combined(plan) + if any((complete.modules, complete.module_secret_keys, + complete.module_secret_owners, complete.builtin_private_keys)): + cfg.set_module_secret_registry( + set(complete.module_secret_keys), dict(complete.module_secret_owners) + ) + cfg.PRIVATE_KEYS.update(complete.builtin_private_keys) + template_loaders, template_dirs = [app.jinja_loader], set() + for contribution in plan.modules: + if contribution.config: + register_module_config( + dict(contribution.config), contribution.module_id, contribution.builtin, + list(contribution.secret_keys), list(contribution.private_keys), + ) + apply_module_i18n(contribution.module_id, { + lang: dict(strings) for lang, strings in contribution.i18n_catalogs + }) + info = contribution.info + info.template_paths = dict(contribution.template_paths) + for name in ( + "collector_class", "publisher_class", "thresholds_data", + "theme_data", "has_css", "has_js", + ): + setattr(info, name, getattr(contribution, name)) + if contribution.thresholds_data is not None: + analyzer.set_thresholds( + contribution.thresholds_data, + profile_id=contribution.module_id, + profile_version=contribution.version, + ) + directory = contribution.template_dir + if directory and directory not in template_dirs: + template_loaders.append(FileSystemLoader(directory)) + template_dirs.add(directory) + if len(template_loaders) > 1: + app.jinja_loader = ChoiceLoader(template_loaders) + app.extensions["docsight_registration_plan"] = complete + app.extensions["docsight_applied_registration_plans"] = (*applied, plan) + + +def apply_plan(app: Flask, plan: RegistrationPlan) -> None: + """Compatibility adapter for the sole registrar.""" + register_plan(app, plan) + + +def build_core_plan() -> RegistrationPlan: + from .blueprints import core_blueprints + from .web import CORE_ROUTES + + return RegistrationPlan( + rules=tuple( + PlannedRule(spec.rule, spec.endpoint, spec.methods, "core", spec.view) + for spec in CORE_ROUTES + ), + blueprints=tuple( + probe_blueprint(blueprint, source="core-blueprint") + for blueprint in core_blueprints() + ), + ) + + +def canonical_manifest(app: Flask, module_loader=None) -> dict[str, object]: + """Return manifest v1 using stable public identifiers only.""" + plan = app.extensions.get("docsight_registration_plan") + sources = {} + if isinstance(plan, RegistrationPlan): + sources.update({(rule.endpoint, rule.rule): rule.source for rule in plan.rules}) + for blueprint in plan.blueprints: + sources.update({ + (rule.endpoint, rule.rule): rule.source for rule in blueprint.rules + }) + routes = sorted(({ + "endpoint": rule.endpoint, + "rule": rule.rule, + "methods": sorted(rule.methods), + "source": sources.get((rule.endpoint, rule.rule), "framework"), + } for rule in app.url_map.iter_rules()), key=lambda item: ( + item["rule"], item["endpoint"], item["methods"] + )) + modules = [] if not hasattr(module_loader, "get_modules") else sorted(({ + "id": module.id, + "version": module.version, + "type": module.type, + "builtin": bool(module.builtin), + "enabled": bool(module.enabled), + "accepted": bool(module.enabled and not module.error), + } for module in module_loader.get_modules()), key=lambda item: item["id"]) + return { + "version": 1, "blueprints": sorted(app.blueprints), + "routes": routes, "modules": modules, + } + + +def manifest_fingerprint(manifest: dict[str, object]) -> str: + payload = json.dumps( + manifest, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() diff --git a/app/web.py b/app/web.py index 2f320c9c..9f4ec94c 100644 --- a/app/web.py +++ b/app/web.py @@ -737,7 +737,7 @@ def inject_auth(): classic_mod = None first_with_data = None for m in theme_modules: - if m.theme_data: + if m.enabled and not m.error and m.theme_data: if first_with_data is None: first_with_data = m if m.id == "docsight.theme_classic": @@ -1640,15 +1640,8 @@ class RouteSpec: } -def register_core_routes(app) -> None: - """Register DOCSight's stable core HTTP surface on one application.""" - for spec in CORE_ROUTES: - app.add_url_rule( - spec.rule, - endpoint=spec.endpoint, - view_func=spec.view, - methods=list(spec.methods), - ) +def install_core_template_hooks(app) -> None: + """Install non-route template and response hooks on one application.""" for name, function in CORE_TEMPLATE_FILTERS.items(): app.add_template_filter(function, name) app.context_processor(inject_browser_url_bootstrap) diff --git a/tests/architecture/test_app_globals_guard.py b/tests/architecture/test_app_globals_guard.py index 8c6f1ee6..8f316f87 100644 --- a/tests/architecture/test_app_globals_guard.py +++ b/tests/architecture/test_app_globals_guard.py @@ -10,6 +10,7 @@ GUARDED_FILES = ( ROOT / "app" / "web.py", ROOT / "app" / "app_factory.py", + ROOT / "app" / "registration.py", ROOT / "app" / "runtime.py", *(ROOT / "app" / "blueprints").glob("*.py"), *(ROOT / "app" / "modules").glob("*/routes.py"), @@ -64,7 +65,11 @@ def test_factory_is_the_only_flask_constructor(): for node in ast.walk(_tree(path)): if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == "Flask": calls.append(path.relative_to(ROOT).as_posix()) - assert calls == ["app/app_factory.py"] + assert sorted(calls) == [ + "app/app_factory.py", + "app/app_factory.py", + "app/registration.py", + ] def test_guarded_modules_have_no_lowercase_state_bindings(): diff --git a/tests/architecture/test_registration_contract.py b/tests/architecture/test_registration_contract.py new file mode 100644 index 00000000..bb2aac27 --- /dev/null +++ b/tests/architecture/test_registration_contract.py @@ -0,0 +1,276 @@ +"""Regression contract for deterministic, atomic application registration.""" + +from __future__ import annotations + +import ast +from dataclasses import FrozenInstanceError +from pathlib import Path +from types import SimpleNamespace + +import pytest +from flask import Blueprint, Flask + +from app.registration import ( + ModuleContribution, + PlannedBlueprint, + PlannedRule, + RegistrationError, + RegistrationPlan, + apply_plan, + probe_blueprint, + register_plan, + validate_plan, +) + + +ROOT = Path(__file__).resolve().parents[2] + + +def _view(): + return "ok" + + +def _rule(rule: str, endpoint: str, source: str = "module:test") -> PlannedRule: + return PlannedRule(rule, endpoint, ("GET",), source, _view) + + +@pytest.mark.parametrize( + ("rules", "match"), + [ + ((_rule("/one", "same", "module:a"), _rule("/two", "same", "module:b")), "endpoint"), + ((_rule("/same", "one", "module:a"), _rule("/same", "two", "module:b")), "route/method"), + ( + ( + _rule("/modules/a/static/", "module_static_a", "module-static:a"), + _rule("/modules/a/static/", "module_static_b", "module-static:b"), + ), + "route/method", + ), + ( + ( + _rule("/modules/a/static/", "module_static_same", "module-static:a"), + _rule("/modules/b/static/", "module_static_same", "module-static:b"), + ), + "endpoint", + ), + ], +) +def test_rule_collisions_are_rejected_before_apply(rules, match): + app = Flask(__name__) + before = tuple(app.url_map.iter_rules()) + + with pytest.raises(RegistrationError, match=match): + validate_plan(RegistrationPlan(rules=rules)) + + assert tuple(app.url_map.iter_rules()) == before + + +def test_duplicate_blueprint_names_are_rejected_before_apply(): + left = Blueprint("duplicate", __name__) + right = Blueprint("duplicate", __name__) + plan = RegistrationPlan( + blueprints=( + PlannedBlueprint("duplicate", "module:a", left), + PlannedBlueprint("duplicate", "module:b", right), + ) + ) + + with pytest.raises(RegistrationError, match="blueprint name"): + validate_plan(plan) + + +def test_blueprint_probe_failure_does_not_mutate_target(): + target = Flask(__name__) + before = ( + tuple(target.url_map.iter_rules()), + dict(target.view_functions), + dict(target.blueprints), + ) + broken = Blueprint("broken", __name__) + + @broken.record + def fail(_state): + raise RuntimeError("registration exploded") + + with pytest.raises(RegistrationError, match="failed preflight"): + probe_blueprint(broken, source="module:broken") + + assert tuple(target.url_map.iter_rules()) == before[0] + assert target.view_functions == before[1] + assert target.blueprints == before[2] + + +def test_probe_preserves_record_once_for_real_registration(): + calls = [] + blueprint = Blueprint("once", __name__) + + @blueprint.record_once + def record_once(state): + calls.append(state.app.name) + + planned = probe_blueprint(blueprint, source="module:once") + target = Flask("registration-target") + apply_plan(target, RegistrationPlan(blueprints=(planned,))) + + assert calls == ["registration.probe", "registration-target"] + + +def test_probe_restores_nested_blueprint_registration_state(): + parent = Blueprint("parent", __name__) + child = Blueprint("child", __name__) + + @parent.cli.command("parent-command") + def parent_command(): + pass + + @child.cli.command("child-command") + def child_command(): + pass + + child.add_url_rule("/nested", "nested", _view) + parent.register_blueprint(child) + cli_names = parent.cli.name, child.cli.name + + planned = probe_blueprint(parent, source="module:nested") + + assert parent._got_registered_once is False + assert child._got_registered_once is False + assert (parent.cli.name, child.cli.name) == cli_names + assert planned.registered_names == ("parent", "parent.child") + + +def test_productive_flask_registration_has_one_owner(): + violations = [] + for path in (ROOT / "app").rglob("*.py"): + if path.name == "registration.py": + continue + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Attribute): + continue + receiver = node.func.value.id if isinstance(node.func.value, ast.Name) else "" + local_blueprint_rule = node.func.attr == "add_url_rule" and ( + receiver in {"bp", "blueprint"} or receiver.endswith("_bp") + ) + if node.func.attr in {"register_blueprint", "add_url_rule"} and not local_blueprint_rule: + violations.append(f"{path.relative_to(ROOT)}:{node.lineno}:{node.func.attr}") + if node.func.attr == "register" and isinstance(node.func.value, ast.Name): + if node.func.value.id in {"bp", "blueprint", "Blueprint"} or node.func.value.id.endswith("_bp"): + violations.append(f"{path.relative_to(ROOT)}:{node.lineno}:Blueprint.register") + assert violations == [] + + +@pytest.mark.parametrize( + ("relative_path", "forbidden"), + [ + ("app/registration.py", "app.module_loader"), + ("app/module_registry.py", "app.registration"), + ], +) +def test_registration_dependencies_have_no_reverse_imports(relative_path, forbidden): + """Keep lower-level registration and discovery independent of the facade.""" + tree = ast.parse((ROOT / relative_path).read_text(encoding="utf-8")) + imported = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + base = f"app.{module}".rstrip(".") if node.level else module + imported.add(base) + if base == "app": + imported.update(f"app.{alias.name}" for alias in node.names) + assert forbidden not in imported + + +def test_manual_builtin_test_registrar_is_removed(): + matches = [] + for path in ROOT.rglob("*.py"): + if path == Path(__file__): + continue + legacy_name = "register_" + "builtin_test_routes" + if legacy_name in path.read_text(encoding="utf-8"): + matches.append(path.relative_to(ROOT).as_posix()) + assert matches == [] + + +def test_factory_calls_one_top_level_registrar(): + tree = ast.parse((ROOT / "app" / "app_factory.py").read_text(encoding="utf-8")) + create_app = next( + node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "create_app" + ) + calls = [ + node.func.id + for node in ast.walk(create_app) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) + and node.func.id in {"register_plan", "apply_plan", "apply_contributions"} + ] + assert calls == ["register_plan"] + + +@pytest.mark.parametrize("collision", ["module_id", "config", "secret"]) +def test_non_http_plan_collisions_are_rejected_without_mutation(collision): + app = Flask(__name__) + before = ( + tuple(app.url_map.iter_rules()), + dict(app.view_functions), + dict(app.blueprints), + dict(app.extensions), + ) + left = ModuleContribution( + "community.left", "community:left", "1.0", False, SimpleNamespace(), + config=(("private_config_name", "private-value"),), + secret_keys=("private_secret_name",), + ) + right = ModuleContribution( + "community.left" if collision == "module_id" else "community.right", + "community:right", "1.0", False, SimpleNamespace(), + config=(("private_config_name" if collision == "config" else "other", "value"),), + secret_keys=(("private_secret_name",) if collision == "secret" else ("other_secret",)), + ) + plan = RegistrationPlan(modules=(left, right)) + + with pytest.raises(RegistrationError) as caught: + register_plan(app, plan) + + assert before == ( + tuple(app.url_map.iter_rules()), + dict(app.view_functions), + dict(app.blueprints), + dict(app.extensions), + ) + assert "private_config_name" not in str(caught.value) + assert "private_secret_name" not in str(caught.value) + + +def test_complete_plan_is_immutable_redacted_and_applied_once(): + from app import config as cfg + + contribution = ModuleContribution( + "community.once", "community:once", "1.0", False, SimpleNamespace(), + config=(("sensitive_config_name", "sensitive-config-value"),), + secret_keys=("sensitive_config_name",), + ) + plan = RegistrationPlan( + modules=(contribution,), + module_secret_keys=("sensitive_config_name",), + module_secret_owners=(("sensitive_config_name", "community.once"),), + ) + encoded = repr(plan) + + assert "sensitive_config_name" not in encoded + assert "sensitive-config-value" not in encoded + with pytest.raises(FrozenInstanceError): + plan.modules = () + + app = Flask(__name__) + apply_once = RegistrationPlan(rules=(_rule("/once", "once"),)) + secret_registry = set(cfg.MODULE_SECRET_KEYS), dict(cfg.MODULE_SECRET_OWNERS) + try: + register_plan(app, apply_once) + after_first = dict(app.extensions) + with pytest.raises(RegistrationError, match="already been applied"): + register_plan(app, apply_once) + assert app.extensions == after_first + finally: + cfg.set_module_secret_registry(*secret_registry) diff --git a/tests/conftest.py b/tests/conftest.py index 94a0e95a..f9362f90 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,5 @@ """Shared factory fixtures for DOCSight tests.""" -import importlib -import json import os from pathlib import Path @@ -9,8 +7,6 @@ from app.app_factory import create_app, default_module_loader_factory from app.config import ConfigManager -from app.builtin_modules import BUILTIN_MODULE_DIRS -from app.module_loader import module_static_endpoint, setup_module_static from app.runtime import DerivedStorageCache, LoginRateLimiter, RuntimeState, get_runtime @@ -36,36 +32,6 @@ def _uses_factory_context(module): return canonical in FACTORY_CONTEXT_MODULES -def register_builtin_test_routes(app): - """Register shipped module route and static contributions on one test app.""" - module_base = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "app", "modules")) - blueprints = set(app.blueprints) - endpoints = {rule.endpoint for rule in app.url_map.iter_rules()} - for module_dir in BUILTIN_MODULE_DIRS: - manifest_path = os.path.join(module_base, module_dir, "manifest.json") - try: - with open(manifest_path, "r", encoding="utf-8") as handle: - manifest = json.load(handle) - except OSError: - continue - contributes = manifest.get("contributes", {}) - if "routes" in contributes: - try: - routes = importlib.import_module(f"app.modules.{module_dir}.routes") - except ImportError: - routes = None - blueprint = getattr(routes, "bp", None) or getattr(routes, "blueprint", None) - if blueprint is not None and blueprint.name not in blueprints: - app.register_blueprint(blueprint) - blueprints.add(blueprint.name) - static_subdir = contributes.get("static") - endpoint = module_static_endpoint(manifest["id"]) - if static_subdir and endpoint not in endpoints: - setup_module_static(app, manifest["id"], os.path.join(module_base, module_dir), static_subdir) - endpoints.add(endpoint) - return app - - @pytest.fixture(autouse=True, scope="module") def _factory_context_for_direct_route_tests(request, tmp_path_factory): """Give direct route-unit modules an isolated factory app and context.""" @@ -73,11 +39,12 @@ def _factory_context_for_direct_route_tests(request, tmp_path_factory): yield return manager = ConfigManager(str(tmp_path_factory.mktemp("factory-context"))) - application = register_builtin_test_routes(create_app( + application = create_app( config_manager=manager, + module_loader_factory=default_module_loader_factory(manager, search_paths=[]), environ={}, testing=True, - )) + ) request.module.app = application with application.app_context(): yield diff --git a/tests/module_loader/test_atomic_registration.py b/tests/module_loader/test_atomic_registration.py new file mode 100644 index 00000000..17f9e093 --- /dev/null +++ b/tests/module_loader/test_atomic_registration.py @@ -0,0 +1,314 @@ +"""Fault-seed tests for all-or-nothing module contribution registration.""" + +from __future__ import annotations + +import json + +import pytest +from flask import Flask + +from app import analyzer, config as cfg +from app.i18n import _TRANSLATIONS +from app.module_loader import ModuleLoader + + +def _manifest( + module_id: str, contributes: dict[str, str], *, config=None, + module_type="analysis", +): + return { + "id": module_id, + "name": module_id, + "description": "atomic registration fixture", + "version": "1.0.0", + "author": "Test", + "minAppVersion": "2026.2", + "type": module_type, + "contributes": contributes, + "config": config or {}, + } + + +def _write_module(root, directory, manifest, routes=None): + module = root / directory + module.mkdir() + (module / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + if routes is not None: + (module / "routes.py").write_text(routes, encoding="utf-8") + return module + + +def _snapshot(app): + return { + "rules": tuple((r.rule, r.endpoint, tuple(sorted(r.methods))) for r in app.url_map.iter_rules()), + "views": dict(app.view_functions), + "blueprints": dict(app.blueprints), + "defaults": dict(cfg.DEFAULTS), + "bool": set(cfg.BOOL_KEYS), + "int": set(cfg.INT_KEYS), + "private": set(cfg.PRIVATE_KEYS), + "module_secrets": set(cfg.MODULE_SECRET_KEYS), + "module_owners": dict(cfg.MODULE_SECRET_OWNERS), + "translations": {lang: dict(values) for lang, values in _TRANSLATIONS.items()}, + "thresholds": analyzer._thresholds, + "threshold_profile": dict(analyzer._threshold_profile), + } + + +def test_mixed_protected_blueprint_is_rejected_without_any_mutation(tmp_path): + _write_module( + tmp_path, + "mixed", + _manifest("community.mixed", {"routes": "routes.py"}), + """ +from flask import Blueprint +bp = Blueprint("mixed_bp", __name__) +bp.add_url_rule("/community-ok", "ok", lambda: "ok") +bp.add_url_rule("/login", "login_shadow", lambda: "bad") +""", + ) + app = Flask(__name__) + before = _snapshot(app) + + loader = ModuleLoader(app, search_paths=[str(tmp_path)]) + module = loader.load_all()[0] + + assert module.error and "protected" in module.error.lower() + assert _snapshot(app) == before + + +def test_blueprint_registration_failure_is_rejected_without_any_mutation(tmp_path): + _write_module( + tmp_path, + "broken", + _manifest("community.broken", {"routes": "routes.py"}), + """ +from flask import Blueprint +bp = Blueprint("broken_bp", __name__) +@bp.record +def fail(state): + raise RuntimeError("registration exploded") +""", + ) + app = Flask(__name__) + before = _snapshot(app) + + module = ModuleLoader(app, search_paths=[str(tmp_path)]).load_all()[0] + + assert module.error and "preflight" in module.error.lower() + assert _snapshot(app) == before + + +def test_late_invalid_threshold_rejects_routes_and_catalogs_atomically(tmp_path): + manifest = _manifest( + "community.late", + { + "routes": "routes.py", + "static": "static/", + "i18n": "i18n/", + "thresholds": "thresholds.json", + }, + config={"community_late_credential": ""}, + ) + manifest["config_secrets"] = ["community_late_credential"] + module = _write_module( + tmp_path, + "late", + manifest, + """ +from flask import Blueprint +bp = Blueprint("late_bp", __name__) +bp.add_url_rule("/community-late", "late", lambda: "late") +""", + ) + (module / "static").mkdir() + (module / "static" / "main.js").write_text("late", encoding="utf-8") + (module / "i18n").mkdir() + (module / "i18n" / "en.json").write_text(json.dumps({"name": "Late"}), encoding="utf-8") + (module / "thresholds.json").write_text(json.dumps({"downstream_power": {}}), encoding="utf-8") + app = Flask(__name__) + before = _snapshot(app) + + info = ModuleLoader(app, search_paths=[str(tmp_path)]).load_all()[0] + + assert info.error and "threshold" in info.error.lower() + assert info.collector_class is None and info.publisher_class is None + assert info.template_paths == {} and info.has_css is False and info.has_js is False + assert _snapshot(app) == before + + +def test_duplicate_plain_config_ownership_rejects_both_modules(tmp_path): + key = "community_shared_setting" + _write_module(tmp_path, "a", _manifest("community.a", {}, config={key: "a"})) + _write_module(tmp_path, "b", _manifest("community.b", {}, config={key: "b"})) + app = Flask(__name__) + before = _snapshot(app) + + modules = ModuleLoader(app, search_paths=[str(tmp_path)]).load_all() + + assert {module.id for module in modules if module.error} == {"community.a", "community.b"} + assert key not in cfg.DEFAULTS + assert _snapshot(app) == before + + +def test_disabled_module_contributes_nothing(tmp_path): + module = _write_module( + tmp_path, + "disabled", + _manifest( + "community.disabled", + {"routes": "routes.py", "static": "static/", "i18n": "i18n/"}, + config={"community_disabled_enabled": True}, + ), + """ +from flask import Blueprint +bp = Blueprint("disabled_bp", __name__) +bp.add_url_rule("/community-disabled", "disabled", lambda: "disabled") +""", + ) + (module / "static").mkdir() + (module / "i18n").mkdir() + (module / "i18n" / "en.json").write_text(json.dumps({"name": "Disabled"}), encoding="utf-8") + app = Flask(__name__) + before = _snapshot(app) + + info = ModuleLoader( + app, + search_paths=[str(tmp_path)], + disabled_ids={"community.disabled"}, + ).load_all()[0] + + assert info.enabled is False and info.error is None + assert _snapshot(app) == before + + +def test_disabled_theme_retains_only_validated_preview_metadata(tmp_path): + theme_data = { + "dark": {"--bg": "#101010", "--accent": "#7654ff"}, + "light": {"--bg": "#fafafa", "--accent": "#5432dd"}, + } + module = _write_module( + tmp_path, + "preview", + _manifest( + "community.preview", {"theme": "theme.json"}, module_type="theme" + ), + ) + (module / "theme.json").write_text(json.dumps(theme_data), encoding="utf-8") + app = Flask(__name__) + before = _snapshot(app) + loader = ModuleLoader( + app, + search_paths=[str(tmp_path)], + disabled_ids={"community.preview"}, + ) + + info = loader.load_all()[0] + + assert info.enabled is False and info.error is None + assert info.theme_data == theme_data + assert info.template_paths == {} + assert info.collector_class is None and info.publisher_class is None + assert info.thresholds_data is None + assert info.has_css is False and info.has_js is False + assert loader.registration_plan == type(loader.registration_plan)() + assert _snapshot(app) == before + + +@pytest.mark.parametrize("action", ["enable", "enable/", "disable", "disable/"]) +def test_community_own_namespace_cannot_shadow_core_actions(tmp_path, action): + _write_module( + tmp_path, + "action-shadow", + _manifest("community.owned", {"routes": "routes.py"}), + f'''\nfrom flask import Blueprint\nbp = Blueprint("owned_action_bp", __name__)\nbp.add_url_rule("/api/modules/community.owned/{action}", "shadow", lambda: "bad")\n''', + ) + + loader = ModuleLoader(Flask(__name__), search_paths=[str(tmp_path)]) + info = loader.load_all()[0] + + assert info.error and "protected route conflicts" in info.error.lower() + assert str(tmp_path) not in info.error + assert loader.registration_plan.blueprints == () + + +def test_community_own_namespace_allows_non_core_action(tmp_path): + _write_module( + tmp_path, + "owned-status", + _manifest("community.owned", {"routes": "routes.py"}), + ''' +from flask import Blueprint +bp = Blueprint("owned_status_bp", __name__) +bp.add_url_rule("/api/modules/community.owned/status", "status", lambda: "ok") +''', + ) + app = Flask(__name__) + loader = ModuleLoader(app, search_paths=[str(tmp_path)]) + + info = loader.load_all()[0] + + assert info.error is None + assert len(loader.registration_plan.blueprints) == 1 + assert app.test_client().get("/api/modules/community.owned/status").data == b"ok" + + +@pytest.mark.parametrize( + ("kind", "spec"), + [ + ("routes", "missing.py"), + ("static", "missing-static/"), + ("i18n", "missing-i18n/"), + ("tab", "templates/missing.html"), + ("collector", "missing.py:MissingCollector"), + ("publisher", "missing.py:MissingPublisher"), + ], +) +def test_explicit_missing_contribution_rejects_module_atomically(tmp_path, kind, spec): + _write_module( + tmp_path, + "missing", + _manifest("community.missing", {kind: spec}), + ) + app = Flask(__name__) + before = _snapshot(app) + + module = ModuleLoader(app, search_paths=[str(tmp_path)]).load_all()[0] + + assert module.error and kind in module.error.lower() + assert _snapshot(app) == before + + +def test_implicit_static_directory_remains_optional(tmp_path): + _write_module(tmp_path, "no-static", _manifest("community.no_static", {})) + app = Flask(__name__) + + module = ModuleLoader(app, search_paths=[str(tmp_path)]).load_all()[0] + + assert module.error is None + assert module.has_css is False and module.has_js is False + + +def test_contribution_failure_diagnostics_are_redacted(tmp_path, caplog): + sensitive_value = "private-token-value-9371" + module = _write_module( + tmp_path, + "sensitive-location", + _manifest( + "community.redacted", + {"collector": "collector.py:SensitiveCollectorName"}, + ), + ) + (module / "collector.py").write_text( + f"raise RuntimeError({(str(module) + ':' + sensitive_value)!r})\n", + encoding="utf-8", + ) + + info = ModuleLoader(Flask(__name__), search_paths=[str(tmp_path)]).load_all()[0] + + diagnostics = caplog.text + "\n" + (info.error or "") + assert "community.redacted" in diagnostics + assert "collector" in diagnostics + assert str(tmp_path) not in diagnostics + assert sensitive_value not in diagnostics + assert "SensitiveCollectorName" not in diagnostics diff --git a/tests/module_loader/test_loading_core.py b/tests/module_loader/test_loading_core.py index 27463b45..58341a11 100644 --- a/tests/module_loader/test_loading_core.py +++ b/tests/module_loader/test_loading_core.py @@ -362,13 +362,16 @@ def test_module_class_loader_keeps_kind_specific_contract(loader, filename, spec cls = loader("test.mod", mod_dir, spec) assert cls is None - assert "class 'Missing" in caplog.text + assert f"{kind} contribution class not found" in caplog.text + assert "MissingCollector" not in caplog.text + assert "MissingPublisher" not in caplog.text caplog.clear() cls = loader("test.bad", "/tmp", filename) assert cls is None - assert f"{kind} spec must be 'file.py:ClassName'" in caplog.text + assert f"invalid {kind} contribution spec" in caplog.text + assert filename not in caplog.text class TestStaticAndTemplates: diff --git a/tests/module_loader/test_security.py b/tests/module_loader/test_security.py index 2e199cfc..b02d2c35 100644 --- a/tests/module_loader/test_security.py +++ b/tests/module_loader/test_security.py @@ -91,7 +91,8 @@ def test_traversal_blocked_on_load(self, tmp_path): mod = next(m for m in loader.get_modules() if m.id == "test.eviltheme") assert mod.error is not None - assert "unsafe manifest reference" in mod.error.lower() + assert "reference is unsafe" in mod.error.lower() + assert "stolen.json" not in mod.error assert mod.theme_data is None def test_slash_in_filename_blocked(self, tmp_path): @@ -119,7 +120,8 @@ def test_slash_in_filename_blocked(self, tmp_path): mod = next(m for m in loader.get_modules() if m.id == "test.slashtheme") assert mod.error is not None - assert "unsafe manifest reference" in mod.error.lower() + assert "reference is unsafe" in mod.error.lower() + assert "subdir/theme.json" not in mod.error def test_valid_theme_filename_works(self, tmp_path): """A well-formed theme filename still loads correctly.""" @@ -146,7 +148,7 @@ def test_valid_theme_filename_works(self, tmp_path): assert mod.theme_data is not None assert "--bg" in mod.theme_data["dark"] - def test_disabled_theme_traversal_blocked(self, tmp_path): + def test_disabled_theme_traversal_blocked(self, tmp_path, monkeypatch, caplog): """Disabled themes with traversal in contributes.theme must also be blocked.""" mod_dir = tmp_path / "disabledevil" mod_dir.mkdir() @@ -164,6 +166,15 @@ def test_disabled_theme_traversal_blocked(self, tmp_path): "type": "theme", "contributes": {"theme": "../stolen_theme.json"}, })) + real_open = open + outside_reads = [] + + def tracking_open(file, *args, **kwargs): + if os.path.realpath(os.fspath(file)) == os.path.realpath(escape_target): + outside_reads.append(file) + return real_open(file, *args, **kwargs) + + monkeypatch.setattr("builtins.open", tracking_open) app = Flask(__name__) app.config["TESTING"] = True @@ -175,7 +186,15 @@ def test_disabled_theme_traversal_blocked(self, tmp_path): loader.load_all() mod = next(m for m in loader.get_modules() if m.id == "test.disabledevil") + diagnostics = caplog.text + "\n" + (mod.error or "") + assert mod.enabled is False + assert mod.error and "reference is unsafe" in mod.error.lower() assert mod.theme_data is None + assert outside_reads == [] + assert str(tmp_path) not in diagnostics + assert "stolen_theme.json" not in diagnostics + assert "--bg" not in diagnostics + assert loader.registration_plan.modules == () class TestContributesPathTraversal: @@ -239,7 +258,8 @@ def test_thresholds_traversal_blocked(self, tmp_path): mod = next(m for m in loader.get_modules() if m.id == "test.evilmod") assert mod.error is not None - assert "unsafe manifest reference" in mod.error.lower() + assert "reference is unsafe" in mod.error.lower() + assert "stolen.json" not in mod.error def test_i18n_traversal_blocked(self, tmp_path): """i18n with traversal must be rejected.""" @@ -263,7 +283,8 @@ def test_i18n_traversal_blocked(self, tmp_path): mod = next(m for m in loader.get_modules() if m.id == "test.i18nmod") assert mod.error is not None - assert "unsafe manifest subpath" in mod.error.lower() + assert "i18n contribution reference is unsafe" in mod.error.lower() + assert "../../etc" not in mod.error def test_publisher_traversal_blocked(self, tmp_path): """Publisher spec with traversal filename must be rejected.""" @@ -298,7 +319,8 @@ def test_static_traversal_blocked(self, tmp_path): mod = next(m for m in loader.get_modules() if m.id == "test.staticmod") assert mod.error is not None - assert "unsafe manifest subpath" in mod.error.lower() + assert "static contribution reference is unsafe" in mod.error.lower() + assert "../../../var/www" not in mod.error def test_driver_is_not_valid_contributes(): diff --git a/tests/test_app_factory.py b/tests/test_app_factory.py index dfed4eb9..cc1662fe 100644 --- a/tests/test_app_factory.py +++ b/tests/test_app_factory.py @@ -54,3 +54,36 @@ def test_factory_installs_base_path_then_outer_proxy(tmp_path): assert isinstance(app.wsgi_app.app, BasePathMiddleware) assert isinstance(app.session_interface, RequestScopedCookieSessionInterface) assert app.config["SESSION_COOKIE_SECURE"] is True + + +def test_core_only_factory_preserves_process_module_registries(tmp_path): + from app import config as config_module + + previous = ( + set(config_module.MODULE_SECRET_KEYS), + dict(config_module.MODULE_SECRET_OWNERS), + set(config_module.PRIVATE_KEYS), + ) + secret_key = "preserved_community_secret" + private_key = "preserved_builtin_private" + try: + config_module.set_module_secret_registry( + {secret_key}, {secret_key: "community.preserved"} + ) + config_module.PRIVATE_KEYS.add(private_key) + + create_app( + config_manager=ConfigManager(str(tmp_path / "core-only")), + environ={}, + testing=True, + ) + + assert config_module.MODULE_SECRET_KEYS == {secret_key} + assert config_module.MODULE_SECRET_OWNERS == { + secret_key: "community.preserved" + } + assert private_key in config_module.PRIVATE_KEYS + finally: + config_module.set_module_secret_registry(previous[0], previous[1]) + config_module.PRIVATE_KEYS.clear() + config_module.PRIVATE_KEYS.update(previous[2]) diff --git a/tests/test_builtin_module_registry.py b/tests/test_builtin_module_registry.py index 85c121fa..cec6678e 100644 --- a/tests/test_builtin_module_registry.py +++ b/tests/test_builtin_module_registry.py @@ -5,6 +5,7 @@ import json from pathlib import Path +import pytest from flask import Flask from app.builtin_modules import BUILTIN_MODULE_DIRS, BUILTIN_PYTHON_CONTRIBUTIONS @@ -21,6 +22,7 @@ validate_theme, validate_thresholds, ) +from app.registration import RegistrationError ROOT = Path(__file__).resolve().parents[1] BUILTIN_MODULES_DIR = ROOT / "app" / "modules" @@ -183,6 +185,16 @@ def test_builtin_python_contribution_missing_spec_sets_module_error(monkeypatch) raise AssertionError("missing built-in static contribution spec must fail closed") +def test_builtin_module_duplicate_ids_fail_closed(monkeypatch): + duplicate_dir = BUILTIN_MODULE_DIRS[0] + monkeypatch.setattr( + "app.module_loader.BUILTIN_MODULE_DIRS", (duplicate_dir, duplicate_dir) + ) + + with pytest.raises(RegistrationError, match="Duplicate built-in module id"): + discover_builtin_modules(str(BUILTIN_MODULES_DIR)) + + def test_discover_builtin_theme_modules_uses_static_registry(monkeypatch): """Built-in theme registration must not scan module directories.""" import app.module_loader as module_loader @@ -228,14 +240,23 @@ def test_builtin_theme_registry_preserves_optional_metadata(): assert themes["docsight.theme_tribu"].homepage == "https://github.com/itsDNNS/tribu" -def test_builtin_theme_duplicate_ids_are_skipped(monkeypatch): +def test_builtin_theme_duplicate_ids_fail_closed(monkeypatch): duplicate = dict(BUILTIN_THEMES[0]) monkeypatch.setattr("app.module_loader.BUILTIN_THEMES", (BUILTIN_THEMES[0], duplicate)) - modules = discover_builtin_theme_modules() + with pytest.raises(RegistrationError, match="Duplicate built-in theme id"): + discover_builtin_theme_modules() + + +def test_builtin_threshold_duplicate_ids_fail_closed(monkeypatch): + duplicate = dict(BUILTIN_THRESHOLD_PROFILES[0]) + monkeypatch.setattr( + "app.module_loader.BUILTIN_THRESHOLD_PROFILES", + (BUILTIN_THRESHOLD_PROFILES[0], duplicate), + ) - assert len(modules) == 1 - assert modules[0].id == BUILTIN_THEMES[0]["id"] + with pytest.raises(RegistrationError, match="Duplicate built-in threshold profile id"): + discover_builtin_threshold_modules() def test_full_loader_prevents_community_theme_shadowing_builtin_theme(tmp_path): diff --git a/tests/test_module_integration.py b/tests/test_module_integration.py index 3eeca2cb..7e181699 100644 --- a/tests/test_module_integration.py +++ b/tests/test_module_integration.py @@ -7,6 +7,8 @@ import pytest from flask import Flask, current_app +from app.app_factory import create_app +from app.config import ConfigManager from app.module_loader import ModuleLoader from app.runtime import current_runtime, get_runtime @@ -14,6 +16,19 @@ FIXTURE_DIR = os.path.join(os.path.dirname(__file__), "fixtures") +def _fixture_loader_factory(*, disabled_ids=None): + def build(application): + loader = ModuleLoader( + application, + search_paths=[FIXTURE_DIR], + disabled_ids=disabled_ids, + ) + loader.load_all() + return loader + + return build + + class TestModuleIntegration: """End-to-end test with a real module fixture.""" @@ -54,11 +69,17 @@ def teardown_method(self): from app import web self._runtime.module_loader = None - def test_full_load_cycle(self): + def test_full_load_cycle(self, tmp_path): """Discover -> validate -> load config + i18n + routes -> serve requests.""" - app = Flask(__name__) - loader = ModuleLoader(app, search_paths=[FIXTURE_DIR]) - modules = loader.load_all() + manager = ConfigManager(str(tmp_path / "full-load")) + app = create_app( + config_manager=manager, + module_loader_factory=_fixture_loader_factory(), + environ={}, + testing=True, + ) + loader = get_runtime(app).module_loader + modules = loader.get_modules() # Discovery (fixtures dir contains both test_module and ui_module) assert len(modules) == 2 @@ -89,15 +110,19 @@ def test_full_load_cycle(self): assert data["pong"] is True assert data["module"] == "test.integration" - def test_disable_skips_loading(self): + def test_disable_skips_loading(self, tmp_path): """Disabled modules are discovered but their contributions are not loaded.""" - app = Flask(__name__) - loader = ModuleLoader( - app, - search_paths=[FIXTURE_DIR], - disabled_ids={"test.integration"}, + manager = ConfigManager(str(tmp_path / "disabled-load")) + manager.save({"disabled_modules": "test.integration"}) + app = create_app( + config_manager=manager, + module_loader_factory=_fixture_loader_factory( + disabled_ids={"test.integration"} + ), + environ={}, + testing=True, ) - modules = loader.load_all() + modules = get_runtime(app).module_loader.get_modules() mod = next(m for m in modules if m.id == "test.integration") assert mod.enabled is False diff --git a/tests/test_modules_api.py b/tests/test_modules_api.py index 3ab985a2..04cf0d22 100644 --- a/tests/test_modules_api.py +++ b/tests/test_modules_api.py @@ -293,7 +293,10 @@ def app_with_theme(self, tmp_path): })) config = ConfigManager(str(tmp_path / "config")) - app, loader = _create_module_app(config, [str(tmp_path)]) + config.save({"disabled_modules": "test.theme1"}) + app, loader = _create_module_app( + config, [str(tmp_path)], disabled_ids={"test.theme1"} + ) yield app, loader @@ -305,5 +308,20 @@ def test_get_themes_returns_theme_data(self, app_with_theme): data = resp.get_json() assert len(data) == 1 assert data[0]["id"] == "test.theme1" + assert data[0]["enabled"] is False assert "dark" in data[0]["theme_data"] assert data[0]["theme_data"]["dark"]["--bg"] == "#111" + assert loader.registration_plan.modules == () + + def test_disabled_theme_preview_appears_in_appearance_gallery(self, app_with_theme): + from app import web + + app, _loader = app_with_theme + response = app.test_client().get("/settings") + + assert response.status_code == 200 + assert b'data-theme-id="test.theme1"' in response.data + with app.test_request_context("/settings"): + context = web.inject_auth() + assert [theme.id for theme in context["all_theme_modules"]] == ["test.theme1"] + assert context["active_theme_id"] != "test.theme1" diff --git a/tests/test_registration_manifest.py b/tests/test_registration_manifest.py new file mode 100644 index 00000000..d6bba4c0 --- /dev/null +++ b/tests/test_registration_manifest.py @@ -0,0 +1,54 @@ +"""Canonical registration manifest and fingerprint contract.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys + +from app.app_factory import create_app, default_module_loader_factory +from app.config import ConfigManager +from app.registration import canonical_manifest, manifest_fingerprint + + +def test_manifest_is_canonical_and_redacted(tmp_path): + manager = ConfigManager(str(tmp_path / "private-data-location")) + secret_name = "module_super_secret_name" + secret_value = "module-super-secret-value" + manager.save({secret_name: secret_value}) + app = create_app(config_manager=manager, environ={}, testing=True) + + manifest = canonical_manifest(app, app.extensions.get("docsight_module_loader")) + encoded = json.dumps(manifest, sort_keys=True) + + assert manifest["version"] == 1 + assert manifest_fingerprint(manifest) == app.extensions["docsight_registration_fingerprint"] + assert str(tmp_path) not in encoded + assert "private-data-location" not in encoded + assert secret_name not in encoded + assert secret_value not in encoded + assert "