RoleForge/app/services/os_registry_pull.py

"""OS test-image pull configuration (Docker Hub / Harbor / Nexus) stored in app_config; secrets encrypted at rest."""

from __future__ import annotations

import base64
import hashlib
import json
from typing import Any

import asyncpg
from cryptography.fernet import Fernet

from app.core.config import get_settings

CONFIG_KEY = "os_registry_pull"


def _fernet() -> Fernet:
    secret = (get_settings().app_secret_key or "change-me").encode()
    digest = hashlib.sha256(secret).digest()
    key = base64.urlsafe_b64encode(digest)
    return Fernet(key)


def encrypt_registry_secret(plain: str | None) -> str:
    p = (plain or "").strip()
    if not p:
        return ""
    return _fernet().encrypt(p.encode()).decode()


def decrypt_registry_secret(token: str | None) -> str:
    t = (token or "").strip()
    if not t:
        return ""
    try:
        return _fernet().decrypt(t.encode()).decode()
    except Exception:  # noqa: BLE001
        return ""


def _normalize_registry_host(raw: str) -> str:
    s = (raw or "").strip()
    if not s:
        return ""
    s = s.replace("https://", "").replace("http://", "").strip()
    return s.split("/", 1)[0].strip().rstrip("/")


def _repository_prefix(provider: str, registry_host: str, repository_path: str) -> str:
    p = (provider or "hub").strip().lower()
    repo_path = (repository_path or "").strip().strip("/")
    if p in ("hub", "", "docker_hub"):
        return repo_path or "inecs/roleforge"
    host = _normalize_registry_host(registry_host)
    path = "/".join(x.strip() for x in repo_path.replace("\\", "/").split("/") if x.strip())
    if not host or not path:
        return ""
    return f"{host}/{path}"


def _login_server(provider: str, registry_host: str) -> str:
    p = (provider or "hub").strip().lower()
    if p in ("hub", "", "docker_hub"):
        return ""
    return _normalize_registry_host(registry_host)


def parse_stored_os_registry(raw: dict[str, Any] | None) -> dict[str, Any]:
    """Normalize stored JSON into API-friendly dict (no password)."""
    d = dict(raw or {})
    provider = str(d.get("provider") or "hub").strip().lower() or "hub"
    if provider not in ("hub", "harbor", "nexus"):
        provider = "hub"
    registry_host = str(d.get("registry_host") or "").strip()
    repository_path = str(d.get("repository_path") or "").strip()
    username = str(d.get("username") or "").strip()
    pw_ct = str(d.get("password_ciphertext") or "").strip()
    return {
        "provider": provider,
        "registry_host": registry_host,
        "repository_path": repository_path,
        "username": username,
        "password_set": bool(pw_ct),
    }


def stored_row_to_molecule_env(row_val: dict[str, Any] | None) -> dict[str, str]:
    """Build subprocess env vars for Molecule / Ansible docker pulls."""
    d = dict(row_val or {})
    provider = str(d.get("provider") or "hub").strip().lower() or "hub"
    registry_host = str(d.get("registry_host") or "").strip()
    repository_path = str(d.get("repository_path") or "").strip()
    username = str(d.get("username") or "").strip()
    password_plain = decrypt_registry_secret(str(d.get("password_ciphertext") or ""))

    prefix = _repository_prefix(provider, registry_host, repository_path)
    if not prefix:
        prefix = (get_settings().roleforge_os_docker_hub_repository or "").strip() or "inecs/roleforge"

    out: dict[str, str] = {"ROLEFORGE_OS_PULL_REPOSITORY": prefix}

    login_srv = _login_server(provider, registry_host)
    if login_srv:
        out["ROLEFORGE_OS_REGISTRY_LOGIN_SERVER"] = login_srv
    if username:
        out["ROLEFORGE_OS_REGISTRY_USERNAME"] = username
    if password_plain:
        out["ROLEFORGE_OS_REGISTRY_PASSWORD"] = password_plain

    if provider in ("hub", "", "docker_hub"):
        out["ROLEFORGE_OS_DOCKER_HUB_REPOSITORY"] = prefix

    return out


async def fetch_os_registry_molecule_env(conn: asyncpg.Connection) -> dict[str, str]:
    row = await conn.fetchrow("select value from app_config where key = $1", CONFIG_KEY)
    raw: dict[str, Any] | None = None
    if row and row["value"] is not None:
        v = row["value"]
        if isinstance(v, dict):
            raw = dict(v)
        elif isinstance(v, str):
            try:
                raw = json.loads(v)
            except (json.JSONDecodeError, TypeError, ValueError):
                raw = {}
    if not raw:
        hub = (get_settings().roleforge_os_docker_hub_repository or "").strip() or "inecs/roleforge"
        return {"ROLEFORGE_OS_PULL_REPOSITORY": hub, "ROLEFORGE_OS_DOCKER_HUB_REPOSITORY": hub}
    return stored_row_to_molecule_env(raw)


def molecule_env_defaults_from_settings() -> dict[str, str]:
    """When DB row is absent (ephemeral runner): match env file defaults."""
    hub = (get_settings().roleforge_os_docker_hub_repository or "").strip() or "inecs/roleforge"
    return {"ROLEFORGE_OS_PULL_REPOSITORY": hub, "ROLEFORGE_OS_DOCKER_HUB_REPOSITORY": hub}