dp-zp-agent/app/security.py

from __future__ import annotations

import hashlib
import hmac
import os
from tools.sync.wiki_git import expected_branch
from typing import Any

from fastapi import (
    HTTPException,
    Security,
    status,
)
from fastapi.security import (
    APIKeyHeader,
    HTTPAuthorizationCredentials,
    HTTPBearer,
)

MIN_SECRET_LENGTH = 32

SEARCH_API_KEY_HEADER = "X-API-Key"
SYNC_API_KEY_HEADER = "X-API-Key"


TRUE_ENV_VALUES = {
    "1",
    "true",
    "yes",
    "on",
}

FALSE_ENV_VALUES = {
    "0",
    "false",
    "no",
    "off",
}


search_api_key_scheme = APIKeyHeader(
    name=SEARCH_API_KEY_HEADER,
    scheme_name="SearchApiKey",
    auto_error=False,
    description=("API kľúč pre vyhľadávanie v ZP Wiki " "cez hlavičku X-API-Key."),
)

search_bearer_scheme = HTTPBearer(
    scheme_name="SearchBearer",
    auto_error=False,
    description=("Bearer token pre vyhľadávanie v ZP Wiki. " "Používa hodnotu SEARCH_API_KEY."),
)

sync_api_key_scheme = APIKeyHeader(
    name=SYNC_API_KEY_HEADER,
    scheme_name="SyncApiKey",
    auto_error=False,
    description=("API kľúč pre manuálne spustenie " "reindexovania."),
)


def required_environment_value(
    name: str,
) -> str:
    value = os.getenv(
        name,
        "",
    ).strip()

    if not value:
        raise RuntimeError("Chýba povinná environment " f"premenná {name}")

    return value


def validate_secret(
    name: str,
) -> str:
    value = required_environment_value(name)

    if len(value) < MIN_SECRET_LENGTH:
        raise RuntimeError(f"{name} musí mať aspoň " f"{MIN_SECRET_LENGTH} znakov")

    return value


def expected_gitea_repository() -> str:
    value = required_environment_value("EXPECTED_GITEA_REPOSITORY")

    parts = value.split("/")

    if len(parts) != 2:
        raise RuntimeError("EXPECTED_GITEA_REPOSITORY musí mať " "tvar vlastník/repozitár")

    owner, repository = parts

    if not owner or not repository or any(character.isspace() for character in value):
        raise RuntimeError("EXPECTED_GITEA_REPOSITORY musí mať " "tvar vlastník/repozitár")

    return value


def webhook_should_pull_git() -> bool:
    value = (
        os.getenv(
            "WEBHOOK_PULL_GIT",
            "false",
        )
        .strip()
        .casefold()
    )

    if value in TRUE_ENV_VALUES:
        return True

    if value in FALSE_ENV_VALUES:
        return False

    raise RuntimeError("WEBHOOK_PULL_GIT musí byť boolean " "hodnota true/false")


def validate_security_configuration() -> None:
    webhook_secret = validate_secret("WEBHOOK_SECRET")

    sync_api_key = validate_secret("SYNC_API_KEY")

    search_api_key = validate_secret("SEARCH_API_KEY")

    # Každá funkcia musí používať vlastný
    # credential. Jeden uniknutý secret tak
    # neposkytne prístup ku všetkým operáciám.
    if (
        len(
            {
                webhook_secret,
                sync_api_key,
                search_api_key,
            }
        )
        != 3
    ):
        raise RuntimeError(
            "WEBHOOK_SECRET, SYNC_API_KEY " "a SEARCH_API_KEY musia byť " "navzájom rozdielne"
        )

    expected_gitea_repository()
    expected_branch()

    # Validujeme aj voliteľnú boolean
    # konfiguráciu už pri štarte aplikácie.
    webhook_should_pull_git()


def require_search_api_key(
    api_key: str | None = Security(search_api_key_scheme),
    bearer: HTTPAuthorizationCredentials | None = Security(search_bearer_scheme),
) -> None:
    expected = validate_secret("SEARCH_API_KEY")

    supplied_credentials: list[str] = []

    if api_key:
        supplied_credentials.append(api_key)

    if bearer is not None and bearer.scheme.casefold() == "bearer" and bearer.credentials:
        supplied_credentials.append(bearer.credentials)

    valid = any(
        hmac.compare_digest(
            supplied,
            expected,
        )
        for supplied in supplied_credentials
    )

    if not valid:
        raise HTTPException(
            status_code=(status.HTTP_401_UNAUTHORIZED),
            detail=("Neplatný alebo chýbajúci " "API kľúč"),
            headers={
                "WWW-Authenticate": ("Bearer"),
            },
        )


def require_sync_api_key(
    api_key: str | None = Security(sync_api_key_scheme),
) -> None:
    expected = validate_secret("SYNC_API_KEY")

    if not api_key or not hmac.compare_digest(
        api_key,
        expected,
    ):
        raise HTTPException(
            status_code=(status.HTTP_401_UNAUTHORIZED),
            detail=("Neplatný alebo chýbajúci " "API kľúč"),
            headers={
                "WWW-Authenticate": ("ApiKey"),
            },
        )


def verify_gitea_signature(
    raw_body: bytes,
    signature: str | None,
    secret: str,
) -> bool:
    if not signature:
        return False

    supplied = signature.strip().casefold()

    if supplied.startswith("sha256="):
        supplied = supplied.removeprefix("sha256=")

    if len(supplied) != 64:
        return False

    try:
        int(
            supplied,
            16,
        )

    except ValueError:
        return False

    expected = hmac.new(
        secret.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(
        expected,
        supplied,
    )


def repository_name_from_payload(
    payload: dict[str, Any],
) -> str | None:
    repository = payload.get("repository")

    if not isinstance(
        repository,
        dict,
    ):
        return None

    # Pri webhooku akceptujeme iba
    # jednoznačný Gitea full_name:
    #
    # vlastník/repozitár
    #
    # Samotné "name" nestačí, pretože
    # rovnaký názov môže existovať pod
    # rôznymi vlastníkmi.
    value = repository.get("full_name")

    if not isinstance(
        value,
        str,
    ):
        return None

    value = value.strip()

    return value or None


def same_repository(
    actual: str,
    expected: str,
) -> bool:
    return hmac.compare_digest(
        actual.strip().casefold(),
        expected.strip().casefold(),
    )