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