115 lines
4.6 KiB
Python
115 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from collections.abc import Callable, Iterable, Mapping, Sequence
|
|
from dataclasses import dataclass, field
|
|
from importlib import import_module
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, FastAPI
|
|
|
|
from govoplan_core.core.modules import ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.server.fastapi import LifespanFactory
|
|
|
|
ManifestFactory = Callable[[], ModuleManifest]
|
|
RouterEnabled = Callable[[object | None, PlatformRegistry], bool]
|
|
AppConfigurator = Callable[[FastAPI, PlatformRegistry, object | None], None]
|
|
DEFAULT_TRUSTED_IMPORT_PREFIXES = ("govoplan_core", "govoplan_", "app")
|
|
TRUSTED_IMPORT_PREFIXES_ENV = "GOVOPLAN_TRUSTED_IMPORT_PREFIXES"
|
|
_MODULE_PATH_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*$")
|
|
_ATTRIBUTE_PATH_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*(?:\.[A-Za-z][A-Za-z0-9_]*)*$")
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RouterContribution:
|
|
router: APIRouter
|
|
required_module: str | None = None
|
|
enabled: RouterEnabled | None = None
|
|
|
|
def should_include(self, settings: object | None, registry: PlatformRegistry) -> bool:
|
|
if self.required_module and not registry.has_module(self.required_module):
|
|
return False
|
|
if self.enabled is not None and not self.enabled(settings, registry):
|
|
return False
|
|
return True
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class GovoplanServerConfig:
|
|
title: str = "GovOPlaN Server"
|
|
version: str = "0.1.0"
|
|
settings: object | None = None
|
|
enabled_modules: str | Iterable[str] = ("access",)
|
|
api_prefix: str = "/api/v1"
|
|
base_routers: Sequence[APIRouter] = ()
|
|
post_module_routers: Sequence[APIRouter] = ()
|
|
extra_routers: Sequence[RouterContribution] = ()
|
|
lifespan: LifespanFactory | None = None
|
|
cors_origins: Iterable[str] = ()
|
|
manifest_factories: Sequence[ManifestFactory] = ()
|
|
module_context_data: Mapping[str, Any] = field(default_factory=dict)
|
|
app_configurators: Sequence[AppConfigurator] = ()
|
|
|
|
|
|
def import_object(path: str) -> Any:
|
|
module_name, attribute = validate_object_path(path)
|
|
_validate_trusted_import_prefix(module_name, trusted_import_prefixes())
|
|
module = import_module(module_name) # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
|
|
value: Any = module
|
|
for part in attribute.split("."):
|
|
value = getattr(value, part)
|
|
return value
|
|
|
|
|
|
def validate_object_path(path: str) -> tuple[str, str]:
|
|
module_name, separator, attribute = path.strip().partition(":")
|
|
if not separator:
|
|
raise ValueError(f"Object path must use module:attribute syntax: {path!r}")
|
|
if not _MODULE_PATH_RE.fullmatch(module_name):
|
|
raise ValueError(f"Object path module is not a valid Python module path: {path!r}")
|
|
if not _ATTRIBUTE_PATH_RE.fullmatch(attribute) or any(part.startswith("_") for part in attribute.split(".")):
|
|
raise ValueError(f"Object path attribute is not a public attribute path: {path!r}")
|
|
return module_name, attribute
|
|
|
|
|
|
def trusted_import_prefixes() -> tuple[str, ...]:
|
|
configured = os.getenv(TRUSTED_IMPORT_PREFIXES_ENV, "").strip()
|
|
if not configured:
|
|
return DEFAULT_TRUSTED_IMPORT_PREFIXES
|
|
prefixes = tuple(item.strip() for item in configured.split(",") if item.strip())
|
|
return prefixes or DEFAULT_TRUSTED_IMPORT_PREFIXES
|
|
|
|
|
|
def _validate_trusted_import_prefix(module_name: str, prefixes: Iterable[str]) -> None:
|
|
if any(_module_matches_prefix(module_name, prefix) for prefix in prefixes):
|
|
return
|
|
raise ValueError(
|
|
f"Object path module {module_name!r} is outside trusted import prefixes. "
|
|
f"Set {TRUSTED_IMPORT_PREFIXES_ENV} to allow custom deployment config modules."
|
|
)
|
|
|
|
|
|
def _module_matches_prefix(module_name: str, prefix: str) -> bool:
|
|
cleaned = prefix.strip()
|
|
if not cleaned:
|
|
return False
|
|
if cleaned.endswith((".", "_")):
|
|
return module_name.startswith(cleaned)
|
|
return module_name == cleaned or module_name.startswith(f"{cleaned}.")
|
|
|
|
|
|
def load_server_config(path: str | None = None) -> GovoplanServerConfig:
|
|
config_path = path or os.getenv("GOVOPLAN_SERVER_CONFIG") or "govoplan_core.server.default_config:get_server_config"
|
|
try:
|
|
loaded = import_object(config_path)
|
|
except ModuleNotFoundError:
|
|
if path or os.getenv("GOVOPLAN_SERVER_CONFIG"):
|
|
raise
|
|
return GovoplanServerConfig()
|
|
config = loaded() if callable(loaded) else loaded
|
|
if not isinstance(config, GovoplanServerConfig):
|
|
raise TypeError(f"{config_path!r} did not return a GovoplanServerConfig")
|
|
return config
|