Gitea-Tools/webui/project_registry.py

"""Load and validate the web UI project registry (#427)."""

from __future__ import annotations

import json
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any

_FORBIDDEN_EXACT_KEYS = frozenset({
    "token",
    "password",
    "secret",
    "credential",
    "auth",
    "api_key",
    "api-key",
})
_FORBIDDEN_KEY_PREFIXES = ("auth_", "api_key_", "api-key_")
_FORBIDDEN_KEY_SUFFIXES = ("_token", "_secret", "_password", "_credential", "_auth")


def _is_forbidden_key(key: str) -> bool:
    lowered = key.lower()
    if lowered in _FORBIDDEN_EXACT_KEYS:
        return True
    return (
        lowered.startswith(_FORBIDDEN_KEY_PREFIXES)
        or lowered.endswith(_FORBIDDEN_KEY_SUFFIXES)
    )

_REQUIRED_PROJECT_FIELDS = (
    "id",
    "repo_name",
    "gitea_owner",
    "remote_host",
    "default_branch",
    "local_checkout_path",
    "profiles",
    "workflow_paths",
)

_REQUIRED_PROFILE_ROLES = ("author", "reviewer", "reconciler")


@dataclass(frozen=True)
class OnboardingStep:
    id: str
    title: str
    description: str


@dataclass(frozen=True)
class ProjectRecord:
    id: str
    repo_name: str
    gitea_owner: str
    remote_host: str
    default_branch: str
    local_checkout_path: str
    profiles: dict[str, str]
    workflow_paths: dict[str, str]
    schema_paths: dict[str, str]
    onboarding_checklist: tuple[OnboardingStep, ...]


@dataclass(frozen=True)
class ProjectRegistry:
    version: int
    projects: tuple[ProjectRecord, ...]
    source_path: Path


def default_registry_path() -> Path:
    override = os.environ.get("WEBUI_PROJECT_REGISTRY", "").strip()
    if override:
        return Path(override).expanduser().resolve()
    return (Path(__file__).resolve().parent / "data" / "projects.registry.json").resolve()


def _reject_credential_keys(obj: Any, *, path: str = "") -> None:
    if isinstance(obj, dict):
        for key, value in obj.items():
            key_path = f"{path}.{key}" if path else key
            if _is_forbidden_key(key):
                raise ValueError(f"registry must not store credentials ({key_path})")
            _reject_credential_keys(value, path=key_path)
    elif isinstance(obj, list):
        for index, item in enumerate(obj):
            _reject_credential_keys(item, path=f"{path}[{index}]")


def _parse_onboarding(raw: list[dict[str, Any]] | None) -> tuple[OnboardingStep, ...]:
    if not raw:
        return ()
    steps: list[OnboardingStep] = []
    for item in raw:
        steps.append(
            OnboardingStep(
                id=str(item["id"]),
                title=str(item["title"]),
                description=str(item["description"]),
            )
        )
    return tuple(steps)


def _parse_project(raw: dict[str, Any]) -> ProjectRecord:
    missing = [field for field in _REQUIRED_PROJECT_FIELDS if field not in raw]
    if missing:
        raise ValueError(f"project missing required fields: {', '.join(missing)}")

    profiles = raw["profiles"]
    if not isinstance(profiles, dict):
        raise ValueError("profiles must be an object")
    for role in _REQUIRED_PROFILE_ROLES:
        if role not in profiles or not profiles[role]:
            raise ValueError(f"profiles.{role} is required")

    workflow_paths = raw["workflow_paths"]
    if not isinstance(workflow_paths, dict) or not workflow_paths:
        raise ValueError("workflow_paths must be a non-empty object")

    schema_paths = raw.get("schema_paths") or {}
    if not isinstance(schema_paths, dict):
        raise ValueError("schema_paths must be an object when present")

    return ProjectRecord(
        id=str(raw["id"]),
        repo_name=str(raw["repo_name"]),
        gitea_owner=str(raw["gitea_owner"]),
        remote_host=str(raw["remote_host"]),
        default_branch=str(raw["default_branch"]),
        local_checkout_path=str(raw["local_checkout_path"]),
        profiles={role: str(profiles[role]) for role in _REQUIRED_PROFILE_ROLES},
        workflow_paths={key: str(value) for key, value in workflow_paths.items()},
        schema_paths={key: str(value) for key, value in schema_paths.items()},
        onboarding_checklist=_parse_onboarding(raw.get("onboarding_checklist")),
    )


def load_registry(path: Path | None = None) -> ProjectRegistry:
    """Load the versioned project registry from disk."""
    source = (path or default_registry_path()).resolve()
    raw_text = source.read_text(encoding="utf-8")
    payload = json.loads(raw_text)
    if not isinstance(payload, dict):
        raise ValueError("registry root must be an object")

    version = payload.get("version")
    if version != 1:
        raise ValueError(f"unsupported registry version: {version!r}")

    _reject_credential_keys(payload)

    projects_raw = payload.get("projects")
    if not isinstance(projects_raw, list) or not projects_raw:
        raise ValueError("projects must be a non-empty array")

    projects = tuple(_parse_project(item) for item in projects_raw)
    return ProjectRegistry(version=version, projects=projects, source_path=source)


def project_to_dict(project: ProjectRecord) -> dict[str, Any]:
    """Serialize a project for JSON API responses."""
    return {
        "id": project.id,
        "repo_name": project.repo_name,
        "gitea_owner": project.gitea_owner,
        "remote_host": project.remote_host,
        "default_branch": project.default_branch,
        "local_checkout_path": project.local_checkout_path,
        "profiles": dict(project.profiles),
        "workflow_paths": dict(project.workflow_paths),
        "schema_paths": dict(project.schema_paths),
        "onboarding_checklist": [
            {"id": step.id, "title": step.title, "description": step.description}
            for step in project.onboarding_checklist
        ],
    }


def registry_to_dict(registry: ProjectRegistry) -> dict[str, Any]:
    return {
        "version": registry.version,
        "source_path": str(registry.source_path),
        "projects": [project_to_dict(project) for project in registry.projects],
    }


def find_project(registry: ProjectRegistry, project_id: str) -> ProjectRecord | None:
    for project in registry.projects:
        if project.id == project_id:
            return project
    return None