feat(platform): connect owned submissions and real browser flows
This commit is contained in:
@@ -16,9 +16,11 @@ from kf_oj.core.logging import configure_logging
|
||||
from kf_oj.core.settings import Settings
|
||||
from kf_oj.infra.artifacts.repository import ArtifactRepository
|
||||
from kf_oj.infra.artifacts.store import ArtifactStore
|
||||
from kf_oj.infra.coordinator import CoordinatorControl
|
||||
from kf_oj.infra.db.session import build_engine, build_session_factory
|
||||
from kf_oj.modules.identity.service import IdentityService
|
||||
from kf_oj.modules.problems.service import ProblemService
|
||||
from kf_oj.modules.submissions.queries import SubmissionQueries
|
||||
from kf_oj.modules.submissions.service import SubmissionService
|
||||
from kf_oj.modules.submissions.worker_access import WorkerAccess
|
||||
|
||||
@@ -58,12 +60,23 @@ def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
app.state.worker_access = WorkerAccess(
|
||||
app.state.sessions, app.state.problems, artifacts
|
||||
)
|
||||
cancellation = (
|
||||
CoordinatorControl(
|
||||
settings.coordinator_url, settings.coordinator_token.get_secret_value()
|
||||
)
|
||||
if settings.coordinator_url and settings.coordinator_token
|
||||
else None
|
||||
)
|
||||
app.state.submission_queries = SubmissionQueries(
|
||||
app.state.sessions, artifacts, cancellation
|
||||
)
|
||||
yield
|
||||
finally:
|
||||
app.state.identity = None
|
||||
app.state.problems = None
|
||||
app.state.worker_access = None
|
||||
app.state.submissions = None
|
||||
app.state.submission_queries = None
|
||||
if engine is not None:
|
||||
await engine.dispose()
|
||||
if store is not None:
|
||||
@@ -82,6 +95,13 @@ def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
app.state.settings = settings
|
||||
app.add_middleware(BodyLimitMiddleware)
|
||||
|
||||
@app.middleware("http")
|
||||
async def private_responses(request, call_next):
|
||||
response = await call_next(request)
|
||||
if request.url.path.startswith(("/api/v1/auth", "/api/v1/submissions", "/api/v1/internal")):
|
||||
response.headers["Cache-Control"] = "no-store"
|
||||
return response
|
||||
|
||||
@app.exception_handler(RequestValidationError)
|
||||
async def invalid_request(_, __):
|
||||
return JSONResponse({"detail": "invalid request"}, status_code=422)
|
||||
|
||||
@@ -1,12 +1,21 @@
|
||||
from typing import Annotated
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, Response
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
|
||||
|
||||
from kf_oj.api.dependencies import unsafe_principal
|
||||
from kf_oj.contracts.submissions import CreateSubmission, SubmissionCreated
|
||||
from kf_oj.api.dependencies import current_principal, unsafe_principal
|
||||
from kf_oj.contracts.submissions import (
|
||||
CancellationAccepted,
|
||||
CreateSubmission,
|
||||
SubmissionCreated,
|
||||
SubmissionDetail,
|
||||
SubmissionPage,
|
||||
SubmissionSource,
|
||||
)
|
||||
from kf_oj.infra.artifacts.store import ArtifactError
|
||||
from kf_oj.modules.identity.public import Principal
|
||||
from kf_oj.modules.submissions.public import (
|
||||
CancelUnavailable,
|
||||
IdempotencyConflict,
|
||||
SubmissionBusy,
|
||||
SubmissionUnavailable,
|
||||
@@ -41,3 +50,70 @@ async def create(
|
||||
raise HTTPException(422, "submission outside admission policy") from None
|
||||
response.headers["Cache-Control"] = "no-store"
|
||||
return result
|
||||
|
||||
|
||||
def queries(request: Request):
|
||||
service = getattr(request.app.state, "submission_queries", None)
|
||||
if service is None:
|
||||
raise HTTPException(503, "submission service unavailable")
|
||||
return service
|
||||
|
||||
|
||||
@router.get("", response_model=SubmissionPage)
|
||||
async def history(
|
||||
request: Request,
|
||||
principal: Annotated[Principal, Depends(current_principal)],
|
||||
after: Annotated[str | None, Query(max_length=128)] = None,
|
||||
limit: Annotated[int, Query(ge=1, le=100)] = 20,
|
||||
):
|
||||
try:
|
||||
return await queries(request).page(principal, after, limit)
|
||||
except ValueError:
|
||||
raise HTTPException(422, "invalid history request") from None
|
||||
|
||||
|
||||
@router.get("/{submission_id}", response_model=SubmissionDetail)
|
||||
async def detail(
|
||||
submission_id: UUID,
|
||||
request: Request,
|
||||
principal: Annotated[Principal, Depends(current_principal)],
|
||||
):
|
||||
try:
|
||||
return await queries(request).detail(principal, submission_id)
|
||||
except SubmissionUnavailable:
|
||||
raise HTTPException(404, "submission unavailable") from None
|
||||
except ValueError:
|
||||
raise HTTPException(503, "stored result unavailable") from None
|
||||
|
||||
|
||||
@router.get("/{submission_id}/source", response_model=SubmissionSource)
|
||||
async def source(
|
||||
submission_id: UUID,
|
||||
request: Request,
|
||||
principal: Annotated[Principal, Depends(current_principal)],
|
||||
):
|
||||
service = queries(request)
|
||||
try:
|
||||
result = await service.detail(principal, submission_id)
|
||||
data = await service.owned_source(principal, submission_id)
|
||||
return SubmissionSource(
|
||||
submission_id=submission_id, language=result.language, source=data.decode("utf-8")
|
||||
)
|
||||
except SubmissionUnavailable:
|
||||
raise HTTPException(404, "submission unavailable") from None
|
||||
except ValueError:
|
||||
raise HTTPException(503, "stored source unavailable") from None
|
||||
|
||||
|
||||
@router.post("/{submission_id}/cancel", status_code=202, response_model=CancellationAccepted)
|
||||
async def cancel(
|
||||
submission_id: UUID,
|
||||
request: Request,
|
||||
principal: Annotated[Principal, Depends(unsafe_principal)],
|
||||
):
|
||||
try:
|
||||
return await queries(request).cancel(principal, submission_id)
|
||||
except SubmissionUnavailable:
|
||||
raise HTTPException(404, "submission unavailable") from None
|
||||
except CancelUnavailable:
|
||||
raise HTTPException(503, "cancellation service unavailable") from None
|
||||
|
||||
@@ -3,7 +3,16 @@ from typing import Annotated, Literal
|
||||
|
||||
from pydantic import Field, model_validator
|
||||
|
||||
from kf_oj.contracts.judging import MAX_SOURCE_BYTES, Identifier, Language, WireModel
|
||||
from kf_oj.contracts.judging import (
|
||||
MAX_SOURCE_BYTES,
|
||||
CompilationStatus,
|
||||
FailureCode,
|
||||
Identifier,
|
||||
Language,
|
||||
Measurement,
|
||||
Verdict,
|
||||
WireModel,
|
||||
)
|
||||
|
||||
EvaluationState = Literal[
|
||||
"queued", "claimed", "compiling", "running", "cancelling", "finished", "cancelled", "failed"
|
||||
@@ -35,3 +44,49 @@ class SubmissionCreated(WireModel):
|
||||
language: Language
|
||||
state: EvaluationState
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class SubmissionSummary(SubmissionCreated):
|
||||
updated_at: datetime
|
||||
attempts: int
|
||||
verdict: Verdict | None
|
||||
failure_code: FailureCode | None
|
||||
|
||||
|
||||
class PublicCaseResult(WireModel):
|
||||
position: int
|
||||
verdict: Verdict
|
||||
cpu_time_us: Measurement | None
|
||||
memory_peak_bytes: Measurement | None
|
||||
|
||||
|
||||
class PublicJudgement(WireModel):
|
||||
compilation_status: CompilationStatus
|
||||
diagnostics: Annotated[str, Field(max_length=65536)]
|
||||
diagnostics_truncated: bool
|
||||
cpu_time_us: Measurement | None
|
||||
memory_peak_bytes: Measurement | None
|
||||
cleanup_complete: bool
|
||||
cases: tuple[PublicCaseResult, ...]
|
||||
|
||||
|
||||
class SubmissionDetail(SubmissionSummary):
|
||||
result: PublicJudgement | None
|
||||
|
||||
|
||||
class SubmissionPage(WireModel):
|
||||
items: tuple[SubmissionSummary, ...]
|
||||
next_after: str | None
|
||||
|
||||
|
||||
class SubmissionSource(WireModel):
|
||||
submission_id: Identifier
|
||||
language: Language
|
||||
source: Annotated[str, Field(max_length=MAX_SOURCE_BYTES, repr=False)]
|
||||
|
||||
|
||||
class CancellationAccepted(WireModel):
|
||||
submission_id: Identifier
|
||||
evaluation_id: Identifier
|
||||
# Acceptance means Coordinator durably accepted intent, not that processes have stopped.
|
||||
accepted: Literal[True] = True
|
||||
|
||||
@@ -14,17 +14,50 @@ class Settings(BaseSettings):
|
||||
log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO"
|
||||
database_url: SecretStr | None = None
|
||||
amqp_url: SecretStr | None = None
|
||||
browser_origins: tuple[str, ...] = ("http://localhost:5173", "http://localhost:8000")
|
||||
browser_origins: tuple[str, ...] = (
|
||||
"http://localhost:5173",
|
||||
"http://localhost:8000",
|
||||
"http://127.0.0.1:5173",
|
||||
"http://127.0.0.1:8000",
|
||||
)
|
||||
session_lifetime_seconds: int = Field(default=43200, ge=300, le=86400)
|
||||
artifact_root: Path | None = None
|
||||
artifact_max_bytes: int = Field(default=536870912, ge=1, le=16 * 1024**3)
|
||||
artifact_max_files: int = Field(default=4096, ge=1, le=65536)
|
||||
judge_admission: AdmissionPolicy | None = None
|
||||
coordinator_url: str | None = None
|
||||
coordinator_token: SecretStr | None = None
|
||||
|
||||
@model_validator(mode="after")
|
||||
def secure_origins(self):
|
||||
from ipaddress import ip_address
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
if (self.coordinator_url is None) != (self.coordinator_token is None):
|
||||
raise ValueError("both coordinator URL and token required")
|
||||
if self.coordinator_url is not None:
|
||||
parts = urlsplit(self.coordinator_url)
|
||||
try:
|
||||
valid = parts.hostname is not None and ip_address(parts.hostname).is_loopback
|
||||
port = parts.port
|
||||
except ValueError:
|
||||
valid, port = False, None
|
||||
if (
|
||||
not valid
|
||||
or parts.scheme != "http"
|
||||
or port is None
|
||||
or parts.path not in {"", "/"}
|
||||
or parts.username
|
||||
or parts.password
|
||||
or parts.query
|
||||
or parts.fragment
|
||||
):
|
||||
raise ValueError("coordinator must be an explicit numeric loopback HTTP origin")
|
||||
import re
|
||||
|
||||
if not re.fullmatch(r"[0-9a-f]{64}", self.coordinator_token.get_secret_value()):
|
||||
raise ValueError("invalid coordinator service credential")
|
||||
|
||||
if not self.browser_origins or len(self.browser_origins) > 8:
|
||||
raise ValueError("one to eight browser origins required")
|
||||
for origin in self.browser_origins:
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
"""Bounded loopback control adapter; never owns evaluation database writes."""
|
||||
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
class CoordinatorControl:
|
||||
def __init__(self, url: str, token: str):
|
||||
self.url, self.token = url.rstrip("/"), token
|
||||
|
||||
async def cancel(self, evaluation_id: UUID) -> bool:
|
||||
try:
|
||||
async with httpx.AsyncClient(
|
||||
timeout=3, trust_env=False, follow_redirects=False
|
||||
) as client:
|
||||
# Response has no business payload. Do not buffer or log arbitrary remote bodies.
|
||||
async with client.stream(
|
||||
"POST",
|
||||
self.url + "/internal/cancel",
|
||||
headers={"Authorization": "Bearer " + self.token},
|
||||
json={
|
||||
"schema_version": 1,
|
||||
"kind": "judge.evaluation",
|
||||
"evaluation_id": str(evaluation_id),
|
||||
},
|
||||
) as response:
|
||||
return response.status_code == 204
|
||||
except httpx.HTTPError:
|
||||
return False
|
||||
@@ -1,3 +1,15 @@
|
||||
from typing import Protocol
|
||||
from uuid import UUID
|
||||
|
||||
|
||||
class CancelUnavailable(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
class CancellationPort(Protocol):
|
||||
async def cancel(self, evaluation_id: UUID) -> bool: ...
|
||||
|
||||
|
||||
class SubmissionUnavailable(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@@ -1,20 +1,148 @@
|
||||
import base64
|
||||
from datetime import datetime
|
||||
from uuid import UUID
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy import select, tuple_
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||
|
||||
from kf_oj.contracts.judging import MAX_SOURCE_BYTES, ArtifactReference
|
||||
from kf_oj.contracts.judging import MAX_SOURCE_BYTES, ArtifactReference, JudgementReport
|
||||
from kf_oj.contracts.submissions import (
|
||||
CancellationAccepted,
|
||||
PublicCaseResult,
|
||||
PublicJudgement,
|
||||
SubmissionDetail,
|
||||
SubmissionPage,
|
||||
SubmissionSummary,
|
||||
)
|
||||
from kf_oj.infra.artifacts.models import Artifact
|
||||
from kf_oj.infra.artifacts.repository import ArtifactRepository
|
||||
from kf_oj.infra.artifacts.store import ArtifactError
|
||||
from kf_oj.modules.identity.public import Principal
|
||||
from kf_oj.modules.submissions.models import Submission
|
||||
from kf_oj.modules.submissions.public import SubmissionUnavailable
|
||||
from kf_oj.modules.submissions.models import Evaluation, Submission
|
||||
from kf_oj.modules.submissions.public import (
|
||||
CancellationPort,
|
||||
CancelUnavailable,
|
||||
SubmissionUnavailable,
|
||||
)
|
||||
|
||||
|
||||
def cursor_encode(submission: Submission) -> str:
|
||||
return (
|
||||
base64.urlsafe_b64encode(
|
||||
f"{submission.created_at.isoformat()}|{submission.id}".encode("ascii")
|
||||
)
|
||||
.decode("ascii")
|
||||
.rstrip("=")
|
||||
)
|
||||
|
||||
|
||||
def cursor_decode(value: str) -> tuple[datetime, UUID]:
|
||||
try:
|
||||
if not 1 <= len(value) <= 128:
|
||||
raise ValueError
|
||||
date, identifier = (
|
||||
base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True)
|
||||
.decode("ascii")
|
||||
.split("|")
|
||||
)
|
||||
timestamp = datetime.fromisoformat(date)
|
||||
if timestamp.utcoffset() is None:
|
||||
raise ValueError
|
||||
return timestamp, UUID(identifier)
|
||||
except (ValueError, UnicodeError):
|
||||
raise ValueError("invalid history cursor") from None
|
||||
|
||||
|
||||
def summary(submission: Submission, evaluation: Evaluation) -> SubmissionSummary:
|
||||
terminal = evaluation.state in {"finished", "failed", "cancelled"}
|
||||
return SubmissionSummary(
|
||||
submission_id=submission.id,
|
||||
evaluation_id=evaluation.id,
|
||||
problem_version_id=submission.problem_version_id,
|
||||
language=submission.language,
|
||||
state=evaluation.state,
|
||||
created_at=submission.created_at,
|
||||
updated_at=evaluation.updated_at,
|
||||
attempts=evaluation.attempts_count,
|
||||
verdict=evaluation.verdict if terminal else None,
|
||||
failure_code=evaluation.failure_code,
|
||||
)
|
||||
|
||||
|
||||
class SubmissionQueries:
|
||||
def __init__(self, sessions: async_sessionmaker, artifacts: ArtifactRepository):
|
||||
def __init__(
|
||||
self,
|
||||
sessions: async_sessionmaker,
|
||||
artifacts: ArtifactRepository,
|
||||
cancellation: CancellationPort | None = None,
|
||||
):
|
||||
self.sessions, self.artifacts = sessions, artifacts
|
||||
self.cancellation = cancellation
|
||||
|
||||
def owned(self, principal: Principal, submission_id: UUID):
|
||||
return (
|
||||
select(Submission, Evaluation)
|
||||
.join(Evaluation, Evaluation.submission_id == Submission.id)
|
||||
.where(Submission.id == submission_id, Submission.account_id == principal.account_id)
|
||||
)
|
||||
|
||||
async def page(self, principal: Principal, after: str | None, limit: int) -> SubmissionPage:
|
||||
if not 1 <= limit <= 100:
|
||||
raise ValueError("invalid history limit")
|
||||
query = (
|
||||
select(Submission, Evaluation)
|
||||
.join(Evaluation, Evaluation.submission_id == Submission.id)
|
||||
.where(Submission.account_id == principal.account_id)
|
||||
.order_by(Submission.created_at.desc(), Submission.id.desc())
|
||||
.limit(limit + 1)
|
||||
)
|
||||
if after is not None:
|
||||
query = query.where(tuple_(Submission.created_at, Submission.id) < cursor_decode(after))
|
||||
async with self.sessions() as session:
|
||||
rows = (await session.execute(query)).all()
|
||||
return SubmissionPage(
|
||||
items=tuple(summary(s, e) for s, e in rows[:limit]),
|
||||
next_after=cursor_encode(rows[limit - 1][0]) if len(rows) > limit else None,
|
||||
)
|
||||
|
||||
async def detail(self, principal: Principal, submission_id: UUID) -> SubmissionDetail:
|
||||
async with self.sessions() as session:
|
||||
row = (await session.execute(self.owned(principal, submission_id))).one_or_none()
|
||||
if row is None:
|
||||
raise SubmissionUnavailable("submission unavailable")
|
||||
submission, evaluation = row
|
||||
result = None
|
||||
if evaluation.state in {"finished", "failed", "cancelled"} and evaluation.final_report:
|
||||
# A single joined snapshot, strict validation and explicit projection prevent leaking
|
||||
# internal bindings, artifacts, hidden case identifiers or partial attempt reports.
|
||||
report = JudgementReport.model_validate(evaluation.final_report)
|
||||
if str(report.binding.evaluation_id) != str(evaluation.id):
|
||||
raise ValueError("stored result binding invalid")
|
||||
if report.verdict.value != evaluation.verdict:
|
||||
raise ValueError("stored result verdict invalid")
|
||||
result = PublicJudgement(
|
||||
compilation_status=report.compilation.status,
|
||||
diagnostics=report.compilation.diagnostics,
|
||||
diagnostics_truncated=report.compilation.diagnostics_truncated,
|
||||
cpu_time_us=report.run_cpu_time_us,
|
||||
memory_peak_bytes=report.run_memory_peak_bytes,
|
||||
cleanup_complete=report.cleanup_complete,
|
||||
cases=tuple(
|
||||
PublicCaseResult(position=i, **c.model_dump(exclude={"case_id"}))
|
||||
for i, c in enumerate(report.cases)
|
||||
),
|
||||
)
|
||||
return SubmissionDetail(**summary(submission, evaluation).model_dump(), result=result)
|
||||
|
||||
async def cancel(self, principal: Principal, submission_id: UUID) -> CancellationAccepted:
|
||||
async with self.sessions() as session:
|
||||
row = (await session.execute(self.owned(principal, submission_id))).one_or_none()
|
||||
if row is None:
|
||||
raise SubmissionUnavailable("submission unavailable")
|
||||
submission, evaluation = row
|
||||
if self.cancellation is None or not await self.cancellation.cancel(evaluation.id):
|
||||
raise CancelUnavailable("cancellation service unavailable")
|
||||
return CancellationAccepted(submission_id=submission.id, evaluation_id=evaluation.id)
|
||||
|
||||
async def owned_source(self, principal: Principal, submission_id: UUID) -> bytes:
|
||||
async with self.sessions() as session:
|
||||
|
||||
Reference in New Issue
Block a user