This commit is contained in:
Ján Pták 2026-08-14 23:45:32 +02:00
parent 3af3431da4
commit 62dd1819cb

View File

@ -5,7 +5,11 @@ import hmac
import os import os
from typing import Any from typing import Any
from fastapi import HTTPException, Security, status from fastapi import (
HTTPException,
Security,
status,
)
from fastapi.security import ( from fastapi.security import (
APIKeyHeader, APIKeyHeader,
HTTPAuthorizationCredentials, HTTPAuthorizationCredentials,
@ -19,6 +23,21 @@ SEARCH_API_KEY_HEADER = "X-API-Key"
SYNC_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( search_api_key_scheme = APIKeyHeader(
name=SEARCH_API_KEY_HEADER, name=SEARCH_API_KEY_HEADER,
scheme_name="SearchApiKey", scheme_name="SearchApiKey",
@ -87,7 +106,26 @@ def expected_gitea_repository() -> str:
"EXPECTED_GITEA_REPOSITORY" "EXPECTED_GITEA_REPOSITORY"
) )
if "/" not in value: 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( raise RuntimeError(
"EXPECTED_GITEA_REPOSITORY musí mať " "EXPECTED_GITEA_REPOSITORY musí mať "
"tvar vlastník/repozitár" "tvar vlastník/repozitár"
@ -102,26 +140,53 @@ def webhook_should_pull_git() -> bool:
"false", "false",
).strip().casefold() ).strip().casefold()
return value in { if value in TRUE_ENV_VALUES:
"1", return True
"true",
"yes", if value in FALSE_ENV_VALUES:
"on", return False
}
raise RuntimeError(
"WEBHOOK_PULL_GIT musí byť boolean "
"hodnota true/false"
)
def validate_security_configuration() -> None: def validate_security_configuration() -> None:
validate_secret( webhook_secret = validate_secret(
"WEBHOOK_SECRET" "WEBHOOK_SECRET"
) )
validate_secret(
sync_api_key = validate_secret(
"SYNC_API_KEY" "SYNC_API_KEY"
) )
validate_secret(
search_api_key = validate_secret(
"SEARCH_API_KEY" "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_gitea_repository()
# Validujeme aj voliteľnú boolean
# konfiguráciu už pri štarte aplikácie.
webhook_should_pull_git()
def require_search_api_key( def require_search_api_key(
api_key: str | None = Security( api_key: str | None = Security(
@ -272,13 +337,16 @@ def repository_name_from_payload(
): ):
return None return None
value = ( # Pri webhooku akceptujeme iba
repository.get( # jednoznačný Gitea full_name:
"full_name" #
) # vlastník/repozitár
or repository.get( #
"name" # Samotné "name" nestačí, pretože
) # rovnaký názov môže existovať pod
# rôznymi vlastníkmi.
value = repository.get(
"full_name"
) )
if not isinstance( if not isinstance(
@ -297,7 +365,6 @@ def same_repository(
expected: str, expected: str,
) -> bool: ) -> bool:
return hmac.compare_digest( return hmac.compare_digest(
actual.casefold(), actual.strip().casefold(),
expected.casefold(), expected.strip().casefold(),
) )