117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
import re
|
|
from dataclasses import dataclass
|
|
|
|
import jwt
|
|
from fastapi import HTTPException, status
|
|
|
|
from app.config import settings
|
|
|
|
REPO_PATH = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$")
|
|
SAFE_FILE = re.compile(r"^[A-Za-z0-9._-]+$")
|
|
|
|
def _parse_files_claim(raw_files: object) -> tuple[str, ...] | None:
|
|
if raw_files is None:
|
|
return None
|
|
if not isinstance(raw_files, list):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT files claim must be a list of paths",
|
|
)
|
|
if not raw_files:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT files claim must not be empty when present",
|
|
)
|
|
if len(raw_files) > settings.max_manifest_files:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=f"JWT files claim exceeds {settings.max_manifest_files} paths",
|
|
)
|
|
|
|
normalized: list[str] = []
|
|
seen: set[str] = set()
|
|
for path in raw_files:
|
|
if not isinstance(path, str):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT files claim must be a list of paths",
|
|
)
|
|
if not SAFE_FILE.match(path):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT files claim must have safe file names",
|
|
)
|
|
if path in seen:
|
|
continue
|
|
seen.add(path)
|
|
normalized.append(path)
|
|
|
|
return tuple(normalized)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BuildClaims:
|
|
repo: str
|
|
dockerfile: str
|
|
dockerfile_sha256: str
|
|
files: tuple[str, ...] | None = None
|
|
|
|
def uploads_authorized(self) -> bool:
|
|
return bool(self.files)
|
|
|
|
|
|
def verify_build_token(token: str) -> BuildClaims:
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
settings.jwt_secret,
|
|
algorithms=[settings.jwt_algorithm],
|
|
)
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT expired please generate a new one",
|
|
) from exc
|
|
except jwt.InvalidTokenError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid JWT",
|
|
) from exc
|
|
|
|
repo = payload.get("repo")
|
|
dockerfile = payload.get("dockerfile")
|
|
dockerfile_sha256 = payload.get("dockerfile_sha256")
|
|
|
|
if not isinstance(repo, str) or not repo.strip():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT missing repo claim",
|
|
)
|
|
repo = repo.strip()
|
|
if not REPO_PATH.match(repo):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT repo must be owner/name on git.lschaefer.xyz",
|
|
)
|
|
|
|
if not isinstance(dockerfile, str) or not dockerfile.strip():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT missing dockerfile claim",
|
|
)
|
|
dockerfile = dockerfile.strip()
|
|
|
|
if not isinstance(dockerfile_sha256, str) or not dockerfile_sha256.strip():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="JWT missing dockerfile_sha256 claim",
|
|
)
|
|
|
|
files = _parse_files_claim(payload.get("files"))
|
|
|
|
return BuildClaims(
|
|
repo=repo,
|
|
dockerfile=dockerfile,
|
|
dockerfile_sha256=dockerfile_sha256,
|
|
files=files,
|
|
)
|