dp-zp-agent/app/security.py

from __future__ import annotations

import hashlib
import hmac
import os
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"


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"
    )

    if "/" not 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()

    return value in {
        "1",
        "true",
        "yes",
        "on",
    }


def validate_security_configuration() -> None:
    validate_secret(
        "WEBHOOK_SECRET"
    )
    validate_secret(
        "SYNC_API_KEY"
    )
    validate_secret(
        "SEARCH_API_KEY"
    )
    expected_gitea_repository()


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

    value = (
        repository.get(
            "full_name"
        )
        or repository.get(
            "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.casefold(),
        expected.casefold(),
    )