From 996dbc51448b64a5c39358f6bc77fe4607fd2910 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Fri, 9 Oct 2026 01:51:13 +0800 Subject: [PATCH] feat(problems): publish immutable versions and fence artifact access --- Makefile | 2 +- README.md | 12 +- backend/README.md | 7 +- backend/pyproject.toml | 2 + backend/src/kf_oj/api/app.py | 28 + backend/src/kf_oj/api/v1/problems.py | 34 + backend/src/kf_oj/api/v1/worker_artifacts.py | 96 +++ backend/src/kf_oj/contracts/judging.py | 11 + backend/src/kf_oj/contracts/problems.py | 31 + backend/src/kf_oj/core/settings.py | 7 + backend/src/kf_oj/infra/artifacts/README.md | 20 + backend/src/kf_oj/infra/artifacts/__init__.py | 2 +- backend/src/kf_oj/infra/artifacts/cli.py | 48 ++ .../src/kf_oj/infra/artifacts/repository.py | 144 ++++ backend/src/kf_oj/infra/artifacts/store.py | 208 ++++++ backend/src/kf_oj/modules/identity/README.md | 4 +- backend/src/kf_oj/modules/problems/README.md | 20 +- backend/src/kf_oj/modules/problems/cli.py | 49 ++ .../kf_oj/modules/problems/import_format.py | 54 ++ .../src/kf_oj/modules/problems/local_files.py | 48 ++ backend/src/kf_oj/modules/problems/public.py | 27 + .../src/kf_oj/modules/problems/repository.py | 59 ++ backend/src/kf_oj/modules/problems/service.py | 224 ++++++ .../src/kf_oj/modules/submissions/README.md | 14 +- .../src/kf_oj/modules/submissions/public.py | 10 + .../src/kf_oj/modules/submissions/queries.py | 39 + .../kf_oj/modules/submissions/repository.py | 27 + .../modules/submissions/worker_access.py | 111 +++ backend/tests/test_artifacts.py | 204 ++++++ backend/tests/test_problems.py | 262 +++++++ backend/tests/test_worker_artifacts.py | 355 +++++++++ contracts/README.md | 5 + contracts/openapi/platform.v1.json | 685 ++++++++++++++++++ frontend/src/api/generated/schema.d.ts | 368 ++++++++++ judge/crates/contracts/src/judging.rs | 22 + .../contracts/tests/judging_examples.rs | 9 + .../contracts/canonical-manifest.v1.txt | 1 + 37 files changed, 3235 insertions(+), 14 deletions(-) create mode 100644 backend/src/kf_oj/api/v1/problems.py create mode 100644 backend/src/kf_oj/api/v1/worker_artifacts.py create mode 100644 backend/src/kf_oj/contracts/problems.py create mode 100644 backend/src/kf_oj/infra/artifacts/README.md create mode 100644 backend/src/kf_oj/infra/artifacts/cli.py create mode 100644 backend/src/kf_oj/infra/artifacts/repository.py create mode 100644 backend/src/kf_oj/infra/artifacts/store.py create mode 100644 backend/src/kf_oj/modules/problems/cli.py create mode 100644 backend/src/kf_oj/modules/problems/import_format.py create mode 100644 backend/src/kf_oj/modules/problems/local_files.py create mode 100644 backend/src/kf_oj/modules/problems/public.py create mode 100644 backend/src/kf_oj/modules/problems/repository.py create mode 100644 backend/src/kf_oj/modules/problems/service.py create mode 100644 backend/src/kf_oj/modules/submissions/public.py create mode 100644 backend/src/kf_oj/modules/submissions/queries.py create mode 100644 backend/src/kf_oj/modules/submissions/repository.py create mode 100644 backend/src/kf_oj/modules/submissions/worker_access.py create mode 100644 backend/tests/test_artifacts.py create mode 100644 backend/tests/test_problems.py create mode 100644 backend/tests/test_worker_artifacts.py create mode 100644 tests/fixtures/contracts/canonical-manifest.v1.txt diff --git a/Makefile b/Makefile index 83679b0..3abf90f 100644 --- a/Makefile +++ b/Makefile @@ -57,7 +57,7 @@ format: cd frontend && $(PNPM) format api: - cd backend && $(UV) run --locked uvicorn kf_oj.api.app:create_app --factory --reload --host 127.0.0.1 --port 8000 + cd backend && $(UV) run --locked uvicorn kf_oj.api.app:create_app --factory --reload --host 127.0.0.1 --port 8000 --no-proxy-headers --limit-concurrency 32 --backlog 64 frontend: cd frontend && $(PNPM) dev diff --git a/README.md b/README.md index 6d0019a..6cae75c 100644 --- a/README.md +++ b/README.md @@ -3,8 +3,9 @@ 以 Rust 自研判题与 Linux 沙箱、以 Python/FastAPI 提供平台服务的在线评测系统。 前端使用 Vue 3、TypeScript、Element Plus 和 Vite。 -当前版本为 **0.0.1 工程骨架**：平台 API 和协调器提供存活接口，前端可检查实际 API 连接。 -沙箱隔离执行、判题、账号、题库、提交与学习业务尚未实现。 +当前版本为 **0.0.1**：沙箱已有受控隔离、C/C++ 编译及 native 执行；平台已实现真实身份、 +不可变题目、工件及有效租约读取基础。完整提交投递、协调器调度、多测试点评测与前端业务 +接入仍在开发，不能将前端预览成绩当作实际结果。 ## 工程划分 @@ -25,7 +26,8 @@ Python 业务域分为 `identity`、`problems`、`submissions`、`community`、` Rust 分为 `kf-contracts`、`kf-sandbox`、`kf-engine`、`kf-worker`、`kf-coordinator` 五个 crate。 协调器拥有执行状态写入权，判题 Worker 不直接写评测数据库。 -沙箱尚未验证隔离能力，当前执行入口明确拒绝运行程序。 +沙箱执行入口每次核验隔离与限额，任何未验证配置继续拒绝执行。 +用户程序及隔离/超限验收只在独立、整体限额的无业务凭据测试 VM 内运行。 ## 安装与检查 @@ -62,11 +64,11 @@ make sandbox-probe # 只读观察 Linux 环境，不创建隔离资源 ``` 本地数据库与队列的启动步骤见 [部署说明](deploy/README.md)。 -业务表和数据库 revision 尚未建立；迁移统一由后端 Alembic 管理。 +业务表和权限分工由后端 Alembic 唯一管理，见 [迁移与权限说明](backend/migrations/README.md)。 模型服务与第三方插件尚未接入。 开发中的规划文档仅在本地维护，不包含在远端仓库中。 -后续逐步实现沙箱、单机判题、分布式恢复、并行优化和学习功能。 +后续逐步接通单机提交判题，再实现分布式恢复、并行优化和学习功能。 ## 许可证 diff --git a/backend/README.md b/backend/README.md index 27d4b56..f1f8d7d 100644 --- a/backend/README.md +++ b/backend/README.md @@ -15,11 +15,12 @@ make api make worker-check ``` -API 当前只有 `/api/v1/health/live`，不连接数据库或消息队列。 +API 提供存活、真实登录/退出/当前用户、公开题目与有效 Worker 租约的清单/工件读取。 +配置数据库后在 lifespan 内组合模块，配置缺失时业务端口明确不可用，存活接口仍可用。 Worker 目前仅支持 `--check`，默认拒绝进入尚未实现的任务循环。 -日志、配置和迁移入口已建立；业务表、登录、判题投递和模型接入尚未实现。 +提交创建、判题投递/成绩和模型接入尚未实现。账号/题目/工件命令及限制见对应模块 README。 Python 虚拟环境位于 `backend/.venv`，可在 `backend/` 直接使用 `uv sync --locked` 和 `uv run --locked`。 -数据库由 Alembic 唯一管理；业务模型落地后在 `migrations/env.py` 中显式登记元数据。 +数据库由 Alembic 唯一管理；业务模型在 `migrations/env.py` 中显式登记元数据。 不要用 API 启动时的 `create_all` 代替迁移，也不要让 Rust 维护另一套迁移历史。 配置示例在 `.env.example`；执行迁移前复制为 `.env` 并设置实际数据库连接。 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 68a991a..3ff4ab7 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -22,6 +22,8 @@ dependencies = [ [project.scripts] kf-accounts = "kf_oj.modules.identity.cli:main" +kf-problems = "kf_oj.modules.problems.cli:main" +kf-artifacts = "kf_oj.infra.artifacts.cli:main" kf-learning-worker = "kf_oj.workers.learning:main" kf-outbox-publisher = "kf_oj.workers.outbox:main" diff --git a/backend/src/kf_oj/api/app.py b/backend/src/kf_oj/api/app.py index 695658e..180a52a 100644 --- a/backend/src/kf_oj/api/app.py +++ b/backend/src/kf_oj/api/app.py @@ -8,11 +8,17 @@ from starlette.responses import JSONResponse from kf_oj import __version__ from kf_oj.api.body_limits import BodyLimitMiddleware from kf_oj.api.v1.identity import router as identity_router +from kf_oj.api.v1.problems import router as problems_router from kf_oj.api.v1.router import router +from kf_oj.api.v1.worker_artifacts import router as worker_router 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.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.worker_access import WorkerAccess def create_app(settings: Settings | None = None) -> FastAPI: @@ -24,6 +30,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: logger = logging.getLogger("kf_oj.api") logger.info("Platform API started") engine = None + store = None try: if settings.database_url is not None: engine = build_engine(settings) @@ -31,11 +38,30 @@ def create_app(settings: Settings | None = None) -> FastAPI: app.state.identity = IdentityService( app.state.sessions, lifetime_seconds=settings.session_lifetime_seconds ) + artifacts = None + if settings.artifact_root is not None: + store = ArtifactStore( + settings.artifact_root, + max_bytes=settings.artifact_max_bytes, + max_files=settings.artifact_max_files, + ) + artifacts = ArtifactRepository(store) + app.state.problems = ProblemService( + app.state.sessions, artifacts, settings.judge_admission + ) + if artifacts is not None: + app.state.worker_access = WorkerAccess( + app.state.sessions, app.state.problems, artifacts + ) yield finally: app.state.identity = None + app.state.problems = None + app.state.worker_access = None if engine is not None: await engine.dispose() + if store is not None: + store.close() logger.info("Platform API stopped") app = FastAPI( @@ -56,4 +82,6 @@ def create_app(settings: Settings | None = None) -> FastAPI: app.include_router(router, prefix="/api/v1") app.include_router(identity_router, prefix="/api/v1") + app.include_router(problems_router, prefix="/api/v1") + app.include_router(worker_router, prefix="/api/v1") return app diff --git a/backend/src/kf_oj/api/v1/problems.py b/backend/src/kf_oj/api/v1/problems.py new file mode 100644 index 0000000..338e0a9 --- /dev/null +++ b/backend/src/kf_oj/api/v1/problems.py @@ -0,0 +1,34 @@ +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, HTTPException, Query, Request + +from kf_oj.contracts.problems import ProblemDetail, ProblemPage +from kf_oj.modules.problems.public import ProblemUnavailable +from kf_oj.modules.problems.service import ProblemService + +router = APIRouter(prefix="/problems", tags=["problems"]) + + +def service(request: Request) -> ProblemService: + component = getattr(request.app.state, "problems", None) + if component is None: + raise HTTPException(503, "problem service unavailable") + return component + + +@router.get("", response_model=ProblemPage) +async def list_problems( + request: Request, + after: Annotated[str | None, Query(pattern=r"^[a-z0-9][a-z0-9_-]{0,63}$")] = None, + limit: Annotated[int, Query(ge=1, le=100)] = 20, +) -> ProblemPage: + return await service(request).list(after=after, limit=limit) + + +@router.get("/versions/{version_id}", response_model=ProblemDetail) +async def problem_version(version_id: UUID, request: Request) -> ProblemDetail: + try: + return await service(request).detail(version_id) + except ProblemUnavailable: + raise HTTPException(404, "problem version unavailable") from None diff --git a/backend/src/kf_oj/api/v1/worker_artifacts.py b/backend/src/kf_oj/api/v1/worker_artifacts.py new file mode 100644 index 0000000..c4da45c --- /dev/null +++ b/backend/src/kf_oj/api/v1/worker_artifacts.py @@ -0,0 +1,96 @@ +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Header, HTTPException, Request, Response, Security +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer + +from kf_oj.contracts.judging import JobManifest, LeaseBinding +from kf_oj.modules.submissions.public import WorkerAccessDenied, WorkerReadBusy +from kf_oj.modules.submissions.worker_access import WorkerAccess + +router = APIRouter(prefix="/internal/evaluations", tags=["worker artifacts"]) +bearer = HTTPBearer(auto_error=False, scheme_name="WorkerLease") + + +def grant( + evaluation_id: UUID, + attempt_id: UUID, + worker: UUID, + fence: int, + credentials: HTTPAuthorizationCredentials | None, +) -> tuple[LeaseBinding, str]: + if ( + credentials is None + or credentials.scheme.lower() != "bearer" + or len(credentials.credentials) != 64 + ): + raise HTTPException(401, "worker lease required") + try: + return LeaseBinding( + evaluation_id=evaluation_id, + attempt_id=attempt_id, + worker_id=worker, + fencing_version=fence, + ), credentials.credentials + except ValueError: + raise HTTPException(404, "worker grant rejected") from None + + +def component(request: Request) -> WorkerAccess: + access = getattr(request.app.state, "worker_access", None) + if access is None: + raise HTTPException(503, "worker artifact service unavailable") + return access + + +@router.get("/{evaluation_id}/attempts/{attempt_id}/manifest", response_model=JobManifest) +async def manifest( + evaluation_id: UUID, + attempt_id: UUID, + request: Request, + worker: Annotated[UUID, Header(alias="X-Worker-ID")], + fence: Annotated[int, Header(alias="X-Fencing-Version", ge=1)], + credentials: Annotated[HTTPAuthorizationCredentials | None, Security(bearer)], +): + binding, token = grant(evaluation_id, attempt_id, worker, fence, credentials) + try: + payload = await component(request).manifest(binding, token) + except WorkerAccessDenied: + raise HTTPException(404, "worker grant rejected") from None + except WorkerReadBusy: + raise HTTPException(429, "worker reads temporarily limited") from None + return Response(payload, media_type="application/json", headers={"Cache-Control": "no-store"}) + + +@router.get( + "/{evaluation_id}/attempts/{attempt_id}/artifacts/{digest}", + response_class=Response, + responses={ + 200: { + "content": { + "application/octet-stream": {"schema": {"type": "string", "format": "binary"}} + } + } + }, +) +async def artifact( + evaluation_id: UUID, + attempt_id: UUID, + digest: str, + request: Request, + worker: Annotated[UUID, Header(alias="X-Worker-ID")], + fence: Annotated[int, Header(alias="X-Fencing-Version", ge=1)], + credentials: Annotated[HTTPAuthorizationCredentials | None, Security(bearer)], +): + binding, token = grant(evaluation_id, attempt_id, worker, fence, credentials) + try: + payload = await component(request).artifact(binding, token, digest) + except WorkerAccessDenied: + raise HTTPException(404, "worker grant rejected") from None + except WorkerReadBusy: + raise HTTPException(429, "worker reads temporarily limited") from None + return Response( + payload, + media_type="application/octet-stream", + headers={"Cache-Control": "no-store", "X-Content-Type-Options": "nosniff"}, + ) diff --git a/backend/src/kf_oj/contracts/judging.py b/backend/src/kf_oj/contracts/judging.py index 0f3d62b..3851354 100644 --- a/backend/src/kf_oj/contracts/judging.py +++ b/backend/src/kf_oj/contracts/judging.py @@ -162,6 +162,17 @@ class JobManifest(WireModel): return self +def manifest_bytes(manifest: JobManifest) -> bytes: + """Stable UTF-8 bytes for persisted snapshots and their SHA-256 identity.""" + return json.dumps( + manifest.model_dump(mode="json"), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + + class LeaseBinding(WireModel): evaluation_id: Identifier attempt_id: Identifier diff --git a/backend/src/kf_oj/contracts/problems.py b/backend/src/kf_oj/contracts/problems.py new file mode 100644 index 0000000..c406828 --- /dev/null +++ b/backend/src/kf_oj/contracts/problems.py @@ -0,0 +1,31 @@ +from typing import Annotated, Literal + +from pydantic import Field + +from kf_oj.contracts.judging import Identifier, Label, Language, RunLimits, WireModel + + +class PublicSample(WireModel): + encoding: Literal["base64"] = "base64" + input: Annotated[str, Field(max_length=87384)] + output: Annotated[str, Field(max_length=87384)] + + +class ProblemSummary(WireModel): + id: Identifier + slug: Label + version_id: Identifier + revision: int + title: str + languages: tuple[Language, ...] + + +class ProblemDetail(ProblemSummary): + statement: str + samples: tuple[PublicSample, ...] + run_limits: RunLimits + + +class ProblemPage(WireModel): + items: tuple[ProblemSummary, ...] + next_after: Label | None diff --git a/backend/src/kf_oj/core/settings.py b/backend/src/kf_oj/core/settings.py index 2e29df1..166c88d 100644 --- a/backend/src/kf_oj/core/settings.py +++ b/backend/src/kf_oj/core/settings.py @@ -1,8 +1,11 @@ +from pathlib import Path from typing import Literal from pydantic import Field, SecretStr, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict +from kf_oj.contracts.judging import AdmissionPolicy + class Settings(BaseSettings): model_config = SettingsConfigDict(env_prefix="KF_OJ_", env_file=".env", extra="ignore") @@ -13,6 +16,10 @@ class Settings(BaseSettings): amqp_url: SecretStr | None = None browser_origins: tuple[str, ...] = ("http://localhost:5173", "http://localhost: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 @model_validator(mode="after") def secure_origins(self): diff --git a/backend/src/kf_oj/infra/artifacts/README.md b/backend/src/kf_oj/infra/artifacts/README.md new file mode 100644 index 0000000..cc3d5ff --- /dev/null +++ b/backend/src/kf_oj/infra/artifacts/README.md @@ -0,0 +1,20 @@ +# 不可变工件 + +存储根配置于 `KF_OJ_ARTIFACT_ROOT`，由服务账号持有，目录 0700、文件 0600。 +单件至多 64 MiB；默认总字节 512 MiB / 文件 4096，可在有界配置中调整。 +发布采用摘要名称、临时文件、文件/目录 fsync 和跨进程锁；已有工件必须通过长度/摘要核验， +拒绝静默覆盖/修复，普通读取不跟随链接。锁等待最多五秒。临时文件也计入容量。 + +适配器执行已经授权的操作，不自行识别用户、租约或题库。公开样例由 problems 的公开 DTO +提供；个人源码由 submissions 的已验证主体与归属端口读取；隐藏数据只经有效当前租约取得。 +没有匿名工件下载或由文件路径选择的读取接口。 + +发布事务先取得 digest 锁，再持久化字节与元数据，业务引用最后提交；事务回滚最多留下 +无元数据的耐久孤立文件，不留下数据库引用指向未发布字节。回收使用同一 digest 锁， +业务引用的数据库生成 ready 状态及复合外键阻止被引用元数据转为 deleting。 + +`uv run --locked kf-artifacts --limit 100 [--after ]` 执行有限回收批次：恢复 deleting、 +标记超过一天的未引用元数据、删除耐久字节/元数据及超过一天的无元数据文件/临时文件。 +每次扫描至多 100 行；显示游标，后续调用接续，避免固定前页上的保留工件阻塞扫描。 +引用工件不能被回收。文件删除后事务失败仍保持 deleting，下一批可恢复；新引用无法加入。 +需要已迁移数据库和有限平台权限；回收不遍历业务域 ORM，也不删除不可变题目版本。 diff --git a/backend/src/kf_oj/infra/artifacts/__init__.py b/backend/src/kf_oj/infra/artifacts/__init__.py index 38d4a81..3fe3bb8 100644 --- a/backend/src/kf_oj/infra/artifacts/__init__.py +++ b/backend/src/kf_oj/infra/artifacts/__init__.py @@ -1 +1 @@ -"""Reserved for authenticated artifact access and digest verification.""" +"""Digest-addressed immutable storage; business ports enforce authorization.""" diff --git a/backend/src/kf_oj/infra/artifacts/cli.py b/backend/src/kf_oj/infra/artifacts/cli.py new file mode 100644 index 0000000..0cadee3 --- /dev/null +++ b/backend/src/kf_oj/infra/artifacts/cli.py @@ -0,0 +1,48 @@ +import argparse +import asyncio + +from sqlalchemy.exc import SQLAlchemyError + +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.db.session import build_engine, build_session_factory + + +async def run(limit: int, after: str | None) -> None: + settings = Settings() + if settings.artifact_root is None: + raise ValueError("artifact root required") + store = ArtifactStore( + settings.artifact_root, + max_bytes=settings.artifact_max_bytes, + max_files=settings.artifact_max_files, + ) + engine = build_engine(settings) + try: + sessions = build_session_factory(engine) + repository = ArtifactRepository(store) + recovered = await repository.finish_deleting(sessions, limit=limit) + marked, cursor = await repository.collect_metadata(sessions, limit=limit, after=after) + removed = await repository.finish_deleting(sessions, limit=limit) + orphans = await repository.collect_orphans(sessions, limit=limit) + print(f"Recovered {recovered}; marked {marked}; removed {removed}; orphans {orphans}.") + print(f"Next metadata scan cursor: {cursor or 'complete'}") + finally: + await engine.dispose() + store.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description="Collect bounded old unreferenced artifacts") + parser.add_argument("--limit", type=int, default=100, choices=range(1, 101)) + parser.add_argument("--after") + args = parser.parse_args() + try: + asyncio.run(run(args.limit, args.after)) + except (ValueError, OSError, SQLAlchemyError): + parser.exit(1, "Artifact collection rejected; check storage and database configuration.\n") + + +if __name__ == "__main__": + main() diff --git a/backend/src/kf_oj/infra/artifacts/repository.py b/backend/src/kf_oj/infra/artifacts/repository.py new file mode 100644 index 0000000..1af4927 --- /dev/null +++ b/backend/src/kf_oj/infra/artifacts/repository.py @@ -0,0 +1,144 @@ +"""Byte durability precedes metadata. Digest locks span caller reference transactions.""" + +import asyncio +from datetime import timedelta + +from sqlalchemy import func, select, text, update +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from kf_oj.contracts.judging import ArtifactReference +from kf_oj.infra.artifacts.models import Artifact +from kf_oj.infra.artifacts.store import ArtifactError, ArtifactStore + + +async def lock_digest(session: AsyncSession, digest: str) -> None: + await session.execute( + text("SELECT pg_advisory_xact_lock(:key)"), + {"key": int(digest[:16], 16) & ((1 << 63) - 1)}, + ) + + +class ArtifactRepository: + def __init__(self, store: ArtifactStore): + self.store = store + + async def put( + self, session: AsyncSession, data: bytes, expected: ArtifactReference | None = None + ) -> ArtifactReference: + reference = self.store.reference(data) + if expected is not None and reference != expected: + raise ArtifactError("artifact digest or length mismatch") + await lock_digest(session, reference.digest) + existing = await session.get(Artifact, reference.digest) + if existing is not None and ( + existing.state != "ready" or existing.bytes != reference.bytes + ): + raise ArtifactError("artifact metadata unavailable") + await asyncio.to_thread(self.store.put, data, reference) + await session.execute( + insert(Artifact) + .values(digest=reference.digest, bytes=reference.bytes) + .on_conflict_do_nothing(index_elements=[Artifact.digest]) + ) + return reference + + async def read(self, session: AsyncSession, reference: ArtifactReference) -> bytes: + metadata = await session.get(Artifact, reference.digest) + if metadata is None or metadata.state != "ready" or metadata.bytes != reference.bytes: + raise ArtifactError("artifact metadata unavailable") + return await asyncio.to_thread(self.store.read, reference) + + async def check_metadata(self, session: AsyncSession, reference: ArtifactReference) -> None: + metadata = await session.get(Artifact, reference.digest) + if metadata is None or metadata.state != "ready" or metadata.bytes != reference.bytes: + raise ArtifactError("artifact metadata unavailable") + + async def collect_metadata( + self, + sessions: async_sessionmaker, + *, + min_age_seconds: int = 86400, + limit: int = 100, + after: str | None = None, + ) -> tuple[int, str | None]: + if min_age_seconds < 3600 or not 1 <= limit <= 100: + raise ArtifactError("invalid collection budget") + if after is not None: + ArtifactReference(digest=after, bytes=0) + async with sessions() as session: + query = select(Artifact.digest).where( + Artifact.state == "ready", + Artifact.created_at < func.now() - timedelta(seconds=min_age_seconds), + ) + if after is not None: + query = query.where(Artifact.digest > after) + names = list(await session.scalars(query.order_by(Artifact.digest).limit(limit))) + marked = 0 + for digest in names: + async with sessions.begin() as session: + try: + async with session.begin_nested(): + changed = await self.mark_deleting(session, digest) + marked += int(changed) + except IntegrityError: + # Referenced rows are protected by their generated ready-state foreign key. + # No domain ORM imports or reverse dependencies are needed for reclamation. + pass + return marked, names[-1] if len(names) == limit else None + + async def collect_orphans( + self, sessions: async_sessionmaker, *, min_age_seconds: int = 86400, limit: int = 100 + ) -> int: + # Only filesystem orphans are collected here. Referenced metadata is never deleted. + # A durable row cannot become an orphan while another transaction publishes it: + # every writer and collector holds the same transaction-scoped digest lock. + names = await asyncio.to_thread( + self.store.old_digests, min_age_seconds=min_age_seconds, limit=limit + ) + collected = 0 + for digest in names: + async with sessions.begin() as session: + await lock_digest(session, digest) + if await session.get(Artifact, digest) is None: + await asyncio.to_thread(self.store.remove, digest) + collected += 1 + await asyncio.to_thread( + self.store.clean_temporary, min_age_seconds=min_age_seconds, limit=limit + ) + return collected + + async def mark_deleting(self, session: AsyncSession, digest: str) -> bool: + # FK(digest, 'ready') on every business reference rejects this transition if referenced. + await lock_digest(session, digest) + result = await session.scalar( + update(Artifact) + .where(Artifact.digest == digest, Artifact.state == "ready") + .values(state="deleting") + .returning(Artifact.digest) + ) + return result is not None + + async def finish_deleting(self, sessions: async_sessionmaker, *, limit: int = 100) -> int: + if not 1 <= limit <= 100: + raise ArtifactError("invalid collection budget") + async with sessions() as session: + names = list( + await session.scalars( + select(Artifact.digest) + .where(Artifact.state == "deleting") + .order_by(Artifact.digest) + .limit(limit) + ) + ) + count = 0 + for digest in names: + async with sessions.begin() as session: + await lock_digest(session, digest) + metadata = await session.get(Artifact, digest) + if metadata is not None and metadata.state == "deleting": + await asyncio.to_thread(self.store.remove, digest) + await session.delete(metadata) + count += 1 + return count diff --git a/backend/src/kf_oj/infra/artifacts/store.py b/backend/src/kf_oj/infra/artifacts/store.py new file mode 100644 index 0000000..3de80ee --- /dev/null +++ b/backend/src/kf_oj/infra/artifacts/store.py @@ -0,0 +1,208 @@ +"""Private immutable byte store. Authorization is applied before calling this adapter.""" + +import fcntl +import hashlib +import os +import re +import stat +import time +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from uuid import uuid4 + +from kf_oj.contracts.judging import MAX_PACKAGE_BYTES, ArtifactReference + +DIGEST = re.compile(r"[0-9a-f]{64}\Z") +TEMP = re.compile(r"\.tmp-[0-9a-f]{32}\Z") + + +class ArtifactError(ValueError): + """Sanitized artifact failure; never includes paths or private byte content.""" + + +class ArtifactStore: + def __init__(self, root: Path, *, max_bytes: int = 512 * 1024 * 1024, max_files: int = 4096): + if not 1 <= max_bytes <= 16 * 1024**3 or not 1 <= max_files <= 65536: + raise ArtifactError("invalid storage capacity") + root = root.absolute() + root.mkdir(mode=0o700, parents=True, exist_ok=True) + if root.resolve() != root: + raise ArtifactError("storage root must not contain links") + self._directory = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC) + self.max_bytes = max_bytes + self.max_files = max_files + info = os.fstat(self._directory) + if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700: + os.close(self._directory) + raise ArtifactError("storage root must be private and owned") + self._closed = False + + def close(self) -> None: + if not self._closed: + os.close(self._directory) + self._closed = True + + @contextmanager + def exclusive(self) -> Iterator[None]: + # Open a different description for every operation: flock also serializes threads. + descriptor = os.open( + ".lock", + os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, + 0o600, + dir_fd=self._directory, + ) + try: + self._regular(os.fstat(descriptor)) + deadline = time.monotonic() + 5 + while True: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + if time.monotonic() >= deadline: + raise ArtifactError("storage lock deadline exceeded") from None + time.sleep(0.01) + yield + finally: + os.close(descriptor) + + @staticmethod + def _regular(info: os.stat_result) -> None: + if ( + not stat.S_ISREG(info.st_mode) + or info.st_nlink != 1 + or info.st_uid != os.getuid() + or stat.S_IMODE(info.st_mode) != 0o600 + ): + raise ArtifactError("invalid storage entry") + + def _entries(self) -> list[tuple[str, os.stat_result]]: + entries = [] + with os.scandir(self._directory) as directory: + for entry in directory: + if entry.name == ".lock": + continue + if not DIGEST.fullmatch(entry.name) and not TEMP.fullmatch(entry.name): + raise ArtifactError("unexpected storage entry") + info = entry.stat(follow_symlinks=False) + self._regular(info) + entries.append((entry.name, info)) + if len(entries) > self.max_files: + raise ArtifactError("storage file capacity exceeded") + return entries + + @staticmethod + def reference(data: bytes) -> ArtifactReference: + if type(data) is not bytes or len(data) > MAX_PACKAGE_BYTES: + raise ArtifactError("artifact size outside policy") + return ArtifactReference(digest=hashlib.sha256(data).hexdigest(), bytes=len(data)) + + def put(self, data: bytes, expected: ArtifactReference | None = None) -> ArtifactReference: + reference = self.reference(data) + if expected is not None and reference != expected: + raise ArtifactError("artifact digest or length mismatch") + with self.exclusive(): + entries = self._entries() + if any(name == reference.digest for name, _ in entries): + self._read(reference) + return reference + if len(entries) + 1 > self.max_files or ( + sum(info.st_size for _, info in entries) + len(data) > self.max_bytes + ): + raise ArtifactError("storage capacity exhausted") + temporary = ".tmp-" + uuid4().hex + descriptor = os.open( + temporary, + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, + 0o600, + dir_fd=self._directory, + ) + try: + with os.fdopen(descriptor, "wb") as output: + output.write(data) + output.flush() + os.fsync(output.fileno()) + # Locking covers every mutation. Never replace or repair an existing digest. + os.rename( + temporary, + reference.digest, + src_dir_fd=self._directory, + dst_dir_fd=self._directory, + ) + os.fsync(self._directory) + finally: + try: + os.unlink(temporary, dir_fd=self._directory) + except FileNotFoundError: + pass + return reference + + def _read(self, reference: ArtifactReference) -> bytes: + if ( + type(reference.digest) is not str + or not DIGEST.fullmatch(reference.digest) + or type(reference.bytes) is not int + or not 0 <= reference.bytes <= MAX_PACKAGE_BYTES + ): + raise ArtifactError("invalid artifact reference") + try: + descriptor = os.open( + reference.digest, + os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC, + dir_fd=self._directory, + ) + with os.fdopen(descriptor, "rb") as source: + info = os.fstat(source.fileno()) + self._regular(info) + if info.st_size != reference.bytes: + raise ArtifactError("artifact digest or length mismatch") + data = source.read(reference.bytes + 1) + if self.reference(data) != reference: + raise ArtifactError("artifact digest or length mismatch") + return data + except OSError: + raise ArtifactError("artifact unavailable") from None + + def read(self, reference: ArtifactReference) -> bytes: + with self.exclusive(): + return self._read(reference) + + def remove(self, digest: str) -> None: + if not DIGEST.fullmatch(digest): + raise ArtifactError("invalid artifact identifier") + with self.exclusive(): + try: + info = os.stat(digest, dir_fd=self._directory, follow_symlinks=False) + self._regular(info) + os.unlink(digest, dir_fd=self._directory) + os.fsync(self._directory) + except FileNotFoundError: + pass + + def old_digests(self, *, min_age_seconds: int, limit: int) -> tuple[str, ...]: + if not 3600 <= min_age_seconds or not 1 <= limit <= 100: + raise ArtifactError("invalid collection budget") + with self.exclusive(): + cutoff = time.time() - min_age_seconds + return tuple( + name + for name, info in self._entries() + if DIGEST.fullmatch(name) and info.st_mtime < cutoff + )[:limit] + + def clean_temporary(self, *, min_age_seconds: int = 86400, limit: int = 100) -> int: + if min_age_seconds < 3600 or not 1 <= limit <= 100: + raise ArtifactError("invalid collection budget") + with self.exclusive(): + cutoff = time.time() - min_age_seconds + names = [ + name + for name, info in self._entries() + if TEMP.fullmatch(name) and info.st_mtime < cutoff + ][:limit] + for name in names: + os.unlink(name, dir_fd=self._directory) + if names: + os.fsync(self._directory) + return len(names) diff --git a/backend/src/kf_oj/modules/identity/README.md b/backend/src/kf_oj/modules/identity/README.md index c2b5e7e..81b9bcf 100644 --- a/backend/src/kf_oj/modules/identity/README.md +++ b/backend/src/kf_oj/modules/identity/README.md @@ -13,7 +13,9 @@ 登录接口为 `/api/v1/auth/login`，退出 `/logout`，当前主体 `/me`。会话使用随机 256-bit token， 数据库只保存 token / CSRF 摘要；默认 12 小时有效，单账号保留至多 10 个有效会话。 -账号禁用立即拒绝登录和旧会话。请求只采用实际连接 peer，不信任客户端自报身份或转发地址。 +账号禁用立即拒绝登录和旧会话。请求只采用 ASGI 连接 peer，不读取客户端转发地址。 +受控启动使用 `--no-proxy-headers`，避免服务器中间件预先替换 peer；反向代理后的用户共享 +该连接预算，引入可信代理的独立限流策略之前不能直接打开客户端转发头信任。 每 peer 至多 10 次登录/15 分钟，预算通过真实数据库串行更新，失败尝试也消耗预算。 摘要计算最多同时 2 次，超出直接 429；限流表至多 4096 桶且每次有界回收过期桶。 diff --git a/backend/src/kf_oj/modules/problems/README.md b/backend/src/kf_oj/modules/problems/README.md index 9b62478..8e39814 100644 --- a/backend/src/kf_oj/modules/problems/README.md +++ b/backend/src/kf_oj/modules/problems/README.md @@ -1,5 +1,21 @@ # problems 负责题目逻辑身份、不可变版本、导入、标签、限制、发布状态与测试工件元数据。 -跨模块访问通过后续 `public.py`；大文件存取通过 `infra/artifacts` 的适配器。 -隐藏数据与公开样例分别授权。当前未创建题库表或导入接口。 +跨模块通过 `public.py` 的不可变判题投影；大文件存取通过 `infra/artifacts` 的适配器。 +公开题面/显式样例与隐藏输入/答案分别授权。公开 API 不返回工件摘要或存储路径。 + +`uv run --locked kf-problems ` 导入并发布一个新版本。目录必须属于当前操作账号， +权限 0700；`problem.json` 遵循 `import_format.py`，只引用该目录下的单层普通文件名。 +不解压归档、不跟随链接、不读取任意路径、不执行导入内容。默认清单可由 `--manifest` 指定。 +要求配置 `KF_OJ_ARTIFACT_ROOT` 和可信 `KF_OJ_JUDGE_ADMISSION` JSON；同一准入配置应交付给节点， +不能由题目放宽。结构示例来自共享 `judge-admission.valid.json`，不是当前宿主隔离验收证明。 + +清单固定 C17/C++20、工具链/镜像版本、精确字节比较、编译/运行/整个作业预算，并声明按序 +测试点 `{key,input_file,expected_file}` 和最多十对公开样例 `{input_file,output_file}`。 +题面以 `statement_file` 读取 UTF-8。每份样例至多 64 KiB，响应采用 base64 保留二进制数据。 +同一 slug 受事务锁保护，每次发布递增 revision；发布后题面、限额、样例、测试点不可修改。 +旧评测继续绑定旧版本；当前列表指向新版本。工件先持久化，再在同一事务建立引用和发布。 + +`GET /api/v1/problems` 以 slug 游标分页（`after` / `limit`，最多 100 项）， +`GET /api/v1/problems/versions/{uuid}` 读取已发布版本。禁用逻辑题目时其版本对公开读取不可用。 +内部判题投影仅供授权应用编排使用；外部 Worker 读取由 submissions 的有效租约端口控制。 diff --git a/backend/src/kf_oj/modules/problems/cli.py b/backend/src/kf_oj/modules/problems/cli.py new file mode 100644 index 0000000..b8e68ca --- /dev/null +++ b/backend/src/kf_oj/modules/problems/cli.py @@ -0,0 +1,49 @@ +"""Local, controlled import of ordinary files; no archive extraction or program execution.""" + +import argparse +import asyncio +from pathlib import Path + +from sqlalchemy.exc import SQLAlchemyError + +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.db.session import build_engine, build_session_factory +from kf_oj.modules.problems.service import ProblemService + + +async def run(root: Path, manifest: str) -> None: + settings = Settings() + if settings.artifact_root is None or settings.judge_admission is None: + raise ValueError("artifact root and trusted judge admission required") + store = ArtifactStore( + settings.artifact_root, + max_bytes=settings.artifact_max_bytes, + max_files=settings.artifact_max_files, + ) + engine = build_engine(settings) + try: + service = ProblemService( + build_session_factory(engine), ArtifactRepository(store), settings.judge_admission + ) + version_id = await service.import_directory(root, manifest) + print(f"Problem version published: {version_id}") + finally: + await engine.dispose() + store.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description="Import immutable problem version") + parser.add_argument("directory", type=Path) + parser.add_argument("--manifest", default="problem.json") + args = parser.parse_args() + try: + asyncio.run(run(args.directory, args.manifest)) + except (ValueError, OSError, SQLAlchemyError): + parser.exit(1, "Problem import rejected; check controlled files, budgets and database.\n") + + +if __name__ == "__main__": + main() diff --git a/backend/src/kf_oj/modules/problems/import_format.py b/backend/src/kf_oj/modules/problems/import_format.py new file mode 100644 index 0000000..b6236c8 --- /dev/null +++ b/backend/src/kf_oj/modules/problems/import_format.py @@ -0,0 +1,54 @@ +"""Local operator format. Filenames never cross into the judge execution protocol.""" + +from typing import Annotated, Literal, Self + +from pydantic import Field, model_validator + +from kf_oj.contracts.judging import ( + MAX_CASES, + MAX_JOB_WALL_US, + Label, + Language, + RunLimits, + Toolchain, + Version, + WireModel, +) + +FileName = Annotated[str, Field(strict=True, pattern=r"^[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}$")] + + +class ImportCase(WireModel): + key: Label + input_file: FileName + expected_file: FileName + + +class ImportSample(WireModel): + input_file: FileName + output_file: FileName + + +class ProblemImport(WireModel): + schema_version: Version + slug: Label + title: Annotated[str, Field(strict=True, min_length=1, max_length=160)] + statement_file: FileName + languages: Annotated[tuple[Language, ...], Field(min_length=1, max_length=2)] + toolchain: Toolchain + comparison: Literal["exact_bytes"] + compile_limits: RunLimits + run_limits: RunLimits + job_wall_time_us: Annotated[int, Field(strict=True, ge=1000, le=MAX_JOB_WALL_US)] + cases: Annotated[tuple[ImportCase, ...], Field(min_length=1, max_length=MAX_CASES)] + samples: Annotated[tuple[ImportSample, ...], Field(max_length=10)] + + @model_validator(mode="after") + def unique(self) -> Self: + if len(set(self.languages)) != len(self.languages): + raise ValueError("duplicate language") + if len({case.key for case in self.cases}) != len(self.cases): + raise ValueError("duplicate testcase key") + if len({(c.input_file, c.expected_file) for c in self.cases}) != len(self.cases): + raise ValueError("duplicate testcase reference") + return self diff --git a/backend/src/kf_oj/modules/problems/local_files.py b/backend/src/kf_oj/modules/problems/local_files.py new file mode 100644 index 0000000..b267138 --- /dev/null +++ b/backend/src/kf_oj/modules/problems/local_files.py @@ -0,0 +1,48 @@ +import os +import re +import stat +from pathlib import Path + + +class LocalImportFiles: + def __init__(self, root: Path): + root = root.absolute() + if root.resolve() != root: + raise ValueError("import directory must not contain links") + self.descriptor = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + info = os.fstat(self.descriptor) + if info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700: + self.close() + raise ValueError("import directory must be private and owned") + + def close(self): + os.close(self.descriptor) + + def read(self, filename: str, maximum: int) -> bytes: + if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}", filename): + raise ValueError("invalid local filename") + try: + descriptor = os.open( + filename, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=self.descriptor + ) + with os.fdopen(descriptor, "rb") as stream: + before = os.fstat(stream.fileno()) + if ( + not stat.S_ISREG(before.st_mode) + or before.st_nlink != 1 + or before.st_uid != os.getuid() + or before.st_size > maximum + ): + raise ValueError("invalid local input file") + data = stream.read(maximum + 1) + after = os.fstat(stream.fileno()) + if ( + len(data) > maximum + or before.st_size != len(data) + or before.st_mtime_ns != after.st_mtime_ns + or before.st_ctime_ns != after.st_ctime_ns + ): + raise ValueError("local input changed during import") + return data + except OSError: + raise ValueError("local input unavailable") from None diff --git a/backend/src/kf_oj/modules/problems/public.py b/backend/src/kf_oj/modules/problems/public.py new file mode 100644 index 0000000..ecc0a23 --- /dev/null +++ b/backend/src/kf_oj/modules/problems/public.py @@ -0,0 +1,27 @@ +"""Authorized immutable version projection; hidden references stay out of public DTOs.""" + +from dataclasses import dataclass +from typing import Protocol +from uuid import UUID + +from kf_oj.contracts.judging import Language, RunLimits, TestCase, Toolchain + + +class ProblemUnavailable(ValueError): + pass + + +@dataclass(frozen=True) +class JudgingVersion: + version_id: UUID + problem_id: UUID + languages: tuple[Language, ...] + toolchain: Toolchain + compile_limits: RunLimits + run_limits: RunLimits + job_wall_time_us: int + cases: tuple[TestCase, ...] + + +class ProblemPort(Protocol): + async def for_submission(self, session, version_id: UUID) -> JudgingVersion: ... diff --git a/backend/src/kf_oj/modules/problems/repository.py b/backend/src/kf_oj/modules/problems/repository.py new file mode 100644 index 0000000..5bd930f --- /dev/null +++ b/backend/src/kf_oj/modules/problems/repository.py @@ -0,0 +1,59 @@ +import hashlib +from uuid import UUID + +from sqlalchemy import func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from kf_oj.modules.problems.models import Problem, ProblemCase, ProblemVersion + + +async def lock_slug(session: AsyncSession, slug: str) -> None: + # Two-key namespace is disjoint from the one-key artifact advisory locks. + key = int.from_bytes(hashlib.sha256(slug.encode()).digest()[:4], signed=True) + await session.execute(text("SELECT pg_advisory_xact_lock(1262907468,:key)"), {"key": key}) + + +async def lock_problem(session: AsyncSession, slug: str) -> Problem | None: + return await session.scalar(select(Problem).where(Problem.slug == slug).with_for_update()) + + +async def next_revision(session: AsyncSession, problem_id: UUID) -> int: + latest = await session.scalar( + select(func.max(ProblemVersion.revision)).where(ProblemVersion.problem_id == problem_id) + ) + return (latest or 0) + 1 + + +async def public_version(session: AsyncSession, version_id: UUID): + return ( + await session.execute( + select(ProblemVersion, Problem) + .join(Problem, Problem.id == ProblemVersion.problem_id) + .where( + ProblemVersion.id == version_id, + ProblemVersion.published_at.is_not(None), + Problem.enabled.is_(True), + ) + ) + ).one_or_none() + + +async def ordered_cases(session: AsyncSession, version_id: UUID): + return list( + await session.scalars( + select(ProblemCase) + .where(ProblemCase.version_id == version_id) + .order_by(ProblemCase.position) + ) + ) + + +async def list_public(session: AsyncSession, after: str | None, limit: int): + query = ( + select(Problem, ProblemVersion) + .join(ProblemVersion, ProblemVersion.id == Problem.active_version_id) + .where(Problem.enabled.is_(True), ProblemVersion.published_at.is_not(None)) + ) + if after is not None: + query = query.where(Problem.slug > after) + return list(await session.execute(query.order_by(Problem.slug).limit(limit))) diff --git a/backend/src/kf_oj/modules/problems/service.py b/backend/src/kf_oj/modules/problems/service.py new file mode 100644 index 0000000..5e414dd --- /dev/null +++ b/backend/src/kf_oj/modules/problems/service.py @@ -0,0 +1,224 @@ +import base64 +from pathlib import Path +from uuid import UUID, uuid4, uuid5 + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from kf_oj.contracts.judging import ( + MAX_PACKAGE_BYTES, + MAX_SOURCE_BYTES, + AdmissionPolicy, + ArtifactReference, + JobManifest, + Language, + RunLimits, + TestCase, + Toolchain, + decode_wire, +) +from kf_oj.contracts.problems import ProblemDetail, ProblemPage, ProblemSummary, PublicSample +from kf_oj.infra.artifacts.models import Artifact +from kf_oj.infra.artifacts.repository import ArtifactRepository +from kf_oj.modules.problems import repository +from kf_oj.modules.problems.import_format import ProblemImport +from kf_oj.modules.problems.local_files import LocalImportFiles +from kf_oj.modules.problems.models import Problem, ProblemCase, ProblemVersion +from kf_oj.modules.problems.public import JudgingVersion, ProblemUnavailable + + +class ProblemService: + def __init__( + self, + sessions: async_sessionmaker, + artifacts: ArtifactRepository | None = None, + policy: AdmissionPolicy | None = None, + ): + self.sessions, self.artifacts, self.policy = sessions, artifacts, policy + + async def import_directory(self, root: Path, manifest_file: str = "problem.json") -> UUID: + if self.artifacts is None or self.policy is None: + raise ValueError("artifact store and trusted node policy required for import") + files = LocalImportFiles(root) + try: + manifest = decode_wire(ProblemImport, files.read(manifest_file, 256 * 1024)) + statement = files.read(manifest.statement_file, 256 * 1024).decode("utf-8") + if not statement or "\x00" in statement or "\x00" in manifest.title: + raise ValueError("invalid public text") + # Read the complete bounded package before writing anything. Public samples are + # separately declared, at most 64 KiB each, and never inferred from hidden tests. + remaining = MAX_PACKAGE_BYTES - MAX_SOURCE_BYTES + payloads = [] + for case in manifest.cases: + data = files.read(case.input_file, remaining) + remaining -= len(data) + answer = files.read( + case.expected_file, min(remaining, manifest.run_limits.stdout_bytes) + ) + remaining -= len(answer) + payloads.append((data, answer)) + samples = tuple( + PublicSample( + input=base64.b64encode(files.read(s.input_file, 65536)).decode("ascii"), + output=base64.b64encode(files.read(s.output_file, 65536)).decode("ascii"), + ) + for s in manifest.samples + ) + finally: + files.close() + version_id = uuid4() + cases = tuple( + TestCase( + case_id=uuid5(version_id, case.key), + input=self.artifacts.store.reference(data), + expected=self.artifacts.store.reference(answer), + ) + for case, (data, answer) in zip(manifest.cases, payloads, strict=True) + ) + check = JobManifest( + schema_version=1, + kind="judge.manifest", + evaluation_id=uuid4(), + submission_id=uuid4(), + problem_version_id=version_id, + language=manifest.languages[0], + toolchain=manifest.toolchain, + source=ArtifactReference(digest="0" * 64, bytes=MAX_SOURCE_BYTES), + cases=cases, + comparison=manifest.comparison, + compile_limits=manifest.compile_limits, + run_limits=manifest.run_limits, + job_wall_time_us=manifest.job_wall_time_us, + ) + self.policy.admit(check, 1) + async with self.sessions.begin() as session: + # Lock existing and not-yet-existing logical slugs before assigning revisions. + await repository.lock_slug(session, manifest.slug) + problem = await repository.lock_problem(session, manifest.slug) + if problem is None: + problem = Problem(id=uuid4(), slug=manifest.slug) + session.add(problem) + await session.flush() + if not problem.enabled: + raise ProblemUnavailable("problem disabled") + revision = await repository.next_revision(session, problem.id) + session.add( + ProblemVersion( + id=version_id, + problem_id=problem.id, + revision=revision, + title=manifest.title, + statement=statement, + languages=[lang.value for lang in manifest.languages], + toolchain=manifest.toolchain.model_dump(mode="json"), + compile_limits=manifest.compile_limits.model_dump(mode="json"), + run_limits=manifest.run_limits.model_dump(mode="json"), + job_wall_time_us=manifest.job_wall_time_us, + comparison=manifest.comparison, + public_samples=[s.model_dump() for s in samples], + ) + ) + await session.flush() + # Consistent digest lock order avoids deadlocks between overlapping imports/jobs. + unique = { + ref.digest: (data, ref) + for case, (input_bytes, answer_bytes) in zip(cases, payloads, strict=True) + for data, ref in [(input_bytes, case.input), (answer_bytes, case.expected)] + } + for digest in sorted(unique): + data, reference = unique[digest] + await self.artifacts.put(session, data, reference) + for position, case in enumerate(cases): + session.add( + ProblemCase( + id=case.case_id, + version_id=version_id, + position=position, + input_digest=case.input.digest, + expected_digest=case.expected.digest, + ) + ) + await session.flush() + version = await session.get(ProblemVersion, version_id) + version.published_at = await session.scalar(select(func.now())) + problem.active_version_id = version_id + return version_id + + @staticmethod + def _summary(problem, version) -> ProblemSummary: + return ProblemSummary( + id=problem.id, + slug=problem.slug, + version_id=version.id, + revision=version.revision, + title=version.title, + languages=tuple(version.languages), + ) + + async def list(self, *, after: str | None = None, limit: int = 20) -> ProblemPage: + if not 1 <= limit <= 100: + raise ValueError("invalid page limit") + async with self.sessions() as session: + rows = await repository.list_public(session, after, limit + 1) + visible = rows[:limit] + return ProblemPage( + items=tuple(self._summary(*row) for row in visible), + next_after=visible[-1][0].slug if len(rows) > limit else None, + ) + + async def detail(self, version_id: UUID) -> ProblemDetail: + async with self.sessions() as session: + row = await repository.public_version(session, version_id) + if row is None: + raise ProblemUnavailable("problem version unavailable") + version, problem = row + return ProblemDetail( + **self._summary(problem, version).model_dump(), + statement=version.statement, + samples=tuple(PublicSample.model_validate(s) for s in version.public_samples), + run_limits=RunLimits.model_validate(version.run_limits), + ) + + async def for_submission(self, session, version_id: UUID) -> JudgingVersion: + if self.artifacts is None: + raise ProblemUnavailable("artifact service unavailable") + row = await repository.public_version(session, version_id) + if row is None: + raise ProblemUnavailable("problem version unavailable") + version, problem = row + cases = await repository.ordered_cases(session, version.id) + refs = [] + digests = {d for case in cases for d in (case.input_digest, case.expected_digest)} + records = { + a.digest: a + for a in await session.scalars(select(Artifact).where(Artifact.digest.in_(digests))) + } + for case in cases: + input_record = records.get(case.input_digest) + answer_record = records.get(case.expected_digest) + if ( + not input_record + or not answer_record + or input_record.state != "ready" + or answer_record.state != "ready" + ): + raise ProblemUnavailable("problem artifacts unavailable") + refs.append( + TestCase( + case_id=case.id, + input=ArtifactReference(digest=case.input_digest, bytes=input_record.bytes), + expected=ArtifactReference( + digest=case.expected_digest, bytes=answer_record.bytes + ), + ) + ) + return JudgingVersion( + version.id, + problem.id, + tuple(Language(v) for v in version.languages), + Toolchain.model_validate(version.toolchain), + RunLimits.model_validate(version.compile_limits), + RunLimits.model_validate(version.run_limits), + version.job_wall_time_us, + tuple(refs), + ) diff --git a/backend/src/kf_oj/modules/submissions/README.md b/backend/src/kf_oj/modules/submissions/README.md index 28d657c..ed86055 100644 --- a/backend/src/kf_oj/modules/submissions/README.md +++ b/backend/src/kf_oj/modules/submissions/README.md @@ -3,4 +3,16 @@ 负责提交幂等请求、不可变源码引用、创建初始评测、历史查询与学生可见结果。 创建提交、评测和待投递事件必须处于同一事务。 协调器拥有领取后的执行状态与最终结果写入权；本模块不能绕过协议覆盖终态。 -后续 `public.py` 提供授权后的学习证据入口。当前未实现提交接口。 +`queries.py` 提供已验证主体的私有源码读取，其他账号不可读取。 +`worker_access.py` 核对当前 evaluation / attempt / Worker / fencing、未撤销及未过期凭证、 +固定 submission 与版本及任务清单摘要；只允许读取该清单声明的工件。 +内部 API 使用 WorkerLease bearer 和 X-Worker-ID / X-Fencing-Version，不使用用户会话授权。 +允许状态为 claimed / compiling / running；取消请求、终态和旧 attempt 均拒绝。 +字节准备后再次核对当前租约；最多同时两个有限读取，超过直接 429，不排无界等待队列。 + +任务清单为 UTF-8 JSON、对象键递归排序、无额外空白，并以这些确切字节的 SHA-256 标识。 +Python/Rust `manifest_bytes` 与共享字节样例保持一致；Coordinator 必须用同一序列化定义存快照。 +读取凭证与客户端提交的身份分开：身份和归属从实际数据库绑定获得，不接受客户端自报 user ID。 + +当前只实现读取与授权基础。提交创建/幂等/outbox、Coordinator 正式 claim、报告及用户成绩 +查询仍在后续模块阶段，不能把测试数据库 fixture 的租约当作正式领取或实际执行证据。 diff --git a/backend/src/kf_oj/modules/submissions/public.py b/backend/src/kf_oj/modules/submissions/public.py new file mode 100644 index 0000000..67a2d83 --- /dev/null +++ b/backend/src/kf_oj/modules/submissions/public.py @@ -0,0 +1,10 @@ +class SubmissionUnavailable(ValueError): + pass + + +class WorkerAccessDenied(ValueError): + """Never reveal whether the evaluation, attempt or artifact exists.""" + + +class WorkerReadBusy(ValueError): + pass diff --git a/backend/src/kf_oj/modules/submissions/queries.py b/backend/src/kf_oj/modules/submissions/queries.py new file mode 100644 index 0000000..1a3cce4 --- /dev/null +++ b/backend/src/kf_oj/modules/submissions/queries.py @@ -0,0 +1,39 @@ +from uuid import UUID + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker + +from kf_oj.contracts.judging import MAX_SOURCE_BYTES, ArtifactReference +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 + + +class SubmissionQueries: + def __init__(self, sessions: async_sessionmaker, artifacts: ArtifactRepository): + self.sessions, self.artifacts = sessions, artifacts + + async def owned_source(self, principal: Principal, submission_id: UUID) -> bytes: + async with self.sessions() as session: + submission = await session.scalar( + select(Submission).where( + Submission.id == submission_id, Submission.account_id == principal.account_id + ) + ) + if submission is None: + raise SubmissionUnavailable("submission unavailable") + metadata = await session.get(Artifact, submission.source_digest) + if ( + metadata is None + or metadata.state != "ready" + or not 1 <= metadata.bytes <= MAX_SOURCE_BYTES + ): + raise SubmissionUnavailable("submission source unavailable") + reference = ArtifactReference(digest=metadata.digest, bytes=metadata.bytes) + try: + return await self.artifacts.read(session, reference) + except ArtifactError: + raise SubmissionUnavailable("submission source unavailable") from None diff --git a/backend/src/kf_oj/modules/submissions/repository.py b/backend/src/kf_oj/modules/submissions/repository.py new file mode 100644 index 0000000..470d4db --- /dev/null +++ b/backend/src/kf_oj/modules/submissions/repository.py @@ -0,0 +1,27 @@ +from sqlalchemy import func, select + +from kf_oj.contracts.judging import LeaseBinding +from kf_oj.modules.submissions.models import Evaluation, JudgeAttempt, Submission + + +async def leased_submission(session, binding: LeaseBinding, token_hash: str): + return ( + await session.execute( + select(Evaluation, JudgeAttempt, Submission) + .join(JudgeAttempt, JudgeAttempt.id == Evaluation.current_attempt_id) + .join(Submission, Submission.id == Evaluation.submission_id) + .where( + Evaluation.id == binding.evaluation_id, + Evaluation.state.in_(("claimed", "compiling", "running")), + Evaluation.cancel_requested_at.is_(None), + Evaluation.fencing_version == binding.fencing_version, + JudgeAttempt.id == binding.attempt_id, + JudgeAttempt.evaluation_id == Evaluation.id, + JudgeAttempt.worker_id == binding.worker_id, + JudgeAttempt.fencing_version == binding.fencing_version, + JudgeAttempt.lease_token_hash == token_hash, + JudgeAttempt.expires_at > func.clock_timestamp(), + JudgeAttempt.revoked_at.is_(None), + ) + ) + ).one_or_none() diff --git a/backend/src/kf_oj/modules/submissions/worker_access.py b/backend/src/kf_oj/modules/submissions/worker_access.py new file mode 100644 index 0000000..b6de521 --- /dev/null +++ b/backend/src/kf_oj/modules/submissions/worker_access.py @@ -0,0 +1,111 @@ +import asyncio +import hashlib +import hmac +import json +import re +from contextlib import asynccontextmanager + +from sqlalchemy.ext.asyncio import async_sessionmaker + +from kf_oj.contracts.judging import ( + JobManifest, + LeaseBinding, + decode_wire, + manifest_bytes, +) +from kf_oj.infra.artifacts.repository import ArtifactRepository +from kf_oj.infra.artifacts.store import ArtifactError +from kf_oj.modules.problems.public import ProblemPort, ProblemUnavailable +from kf_oj.modules.submissions import repository +from kf_oj.modules.submissions.public import WorkerAccessDenied, WorkerReadBusy + + +class WorkerAccess: + def __init__( + self, sessions: async_sessionmaker, problems: ProblemPort, artifacts: ArtifactRepository + ): + self.sessions, self.problems, self.artifacts = sessions, problems, artifacts + self._reads = asyncio.Semaphore(2) + + @asynccontextmanager + async def _read_slot(self): + if self._reads.locked(): + raise WorkerReadBusy("worker read capacity exhausted") + async with self._reads: + yield + + @staticmethod + def _token_hash(token: str) -> str: + if not re.fullmatch(r"[0-9a-f]{64}", token): + raise WorkerAccessDenied("worker grant rejected") + return hashlib.sha256(token.encode("ascii")).hexdigest() + + async def _manifest(self, session, binding: LeaseBinding, token: str) -> JobManifest: + row = await repository.leased_submission(session, binding, self._token_hash(token)) + if row is None: + raise WorkerAccessDenied("worker grant rejected") + evaluation, attempt, submission = row + # Use a sanitized decoder at this trust boundary. JSONB snapshots are immutable for + # runtime roles; use canonical bytes for both retrieval and digest verification. + + try: + manifest = decode_wire(JobManifest, json.dumps(attempt.manifest).encode()) + if ( + manifest.evaluation_id != evaluation.id + or manifest.submission_id != submission.id + or manifest.problem_version_id != evaluation.problem_version_id + or manifest.problem_version_id != submission.problem_version_id + or manifest.language.value != submission.language + or manifest.source.digest != submission.source_digest + or not hmac.compare_digest( + hashlib.sha256(manifest_bytes(manifest)).hexdigest(), attempt.manifest_digest + ) + ): + raise WorkerAccessDenied("worker snapshot rejected") + version = await self.problems.for_submission(session, manifest.problem_version_id) + await self.artifacts.check_metadata(session, manifest.source) + if ( + version.problem_id != submission.problem_id + or manifest.language not in version.languages + or manifest.toolchain != version.toolchain + or manifest.cases != version.cases + or manifest.compile_limits != version.compile_limits + or manifest.run_limits != version.run_limits + or manifest.job_wall_time_us != version.job_wall_time_us + ): + raise WorkerAccessDenied("worker snapshot rejected") + return manifest + except (ValueError, ProblemUnavailable): + raise WorkerAccessDenied("worker snapshot rejected") from None + + async def manifest(self, binding: LeaseBinding, token: str) -> bytes: + async with self._read_slot(), self.sessions() as session: + manifest = await self._manifest(session, binding, token) + payload = manifest_bytes(manifest) + # READ COMMITTED: recheck after all preparation and just before releasing bytes. + if ( + await repository.leased_submission(session, binding, self._token_hash(token)) + is None + ): + raise WorkerAccessDenied("worker grant rejected") + return payload + + async def artifact(self, binding: LeaseBinding, token: str, digest: str) -> bytes: + async with self._read_slot(), self.sessions() as session: + manifest = await self._manifest(session, binding, token) + references = (manifest.source,) + tuple( + reference for case in manifest.cases for reference in (case.input, case.expected) + ) + reference = next((ref for ref in references if ref.digest == digest), None) + if reference is None: + raise WorkerAccessDenied("artifact grant rejected") + try: + payload = await self.artifacts.read(session, reference) + except ArtifactError: + raise WorkerAccessDenied("artifact unavailable") from None + if ( + await repository.leased_submission(session, binding, self._token_hash(token)) + is None + ): + raise WorkerAccessDenied("worker grant rejected") + return payload diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py new file mode 100644 index 0000000..2a3fdd8 --- /dev/null +++ b/backend/tests/test_artifacts.py @@ -0,0 +1,204 @@ +import asyncio +import hashlib +import os +from uuid import uuid4 + +import pytest +from sqlalchemy import func, select, text, update +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from kf_oj.contracts.judging import ArtifactReference +from kf_oj.infra.artifacts.models import Artifact +from kf_oj.infra.artifacts.repository import ArtifactRepository +from kf_oj.infra.artifacts.store import ArtifactError, ArtifactStore + + +def test_immutable_store_digest_capacity_and_recovery(tmp_path): + root = tmp_path / "artifacts" + store = ArtifactStore(root, max_bytes=6, max_files=2) + try: + first = store.put(b"abc") + assert store.read(first) == b"abc" + assert store.put(b"abc") == first + with pytest.raises(ArtifactError, match="digest or length"): + store.put(b"changed", first) + second = store.put(b"def") + with pytest.raises(ArtifactError, match="capacity"): + store.put(b"g") + assert store.read(first) == b"abc" and store.read(second) == b"def" + (root / first.digest).write_bytes(b"xyz") + with pytest.raises(ArtifactError, match="digest or length"): + store.read(first) + with pytest.raises(ArtifactError, match="digest or length"): + store.put(b"abc") # Corrupt files are not silently repaired or overwritten. + assert (root / first.digest).read_bytes() == b"xyz" + finally: + store.close() + + +@pytest.mark.parametrize("attack", ["symlink", "hardlink", "fifo", "traversal"]) +def test_links_and_non_regular_entries_refused(tmp_path, attack): + root = tmp_path / "artifacts" + store = ArtifactStore(root) + data = b"private bytes" + ref = ArtifactReference(digest=hashlib.sha256(data).hexdigest(), bytes=len(data)) + private = tmp_path / "outside" + private.write_bytes(data) + private.chmod(0o600) + try: + if attack == "symlink": + (root / ref.digest).symlink_to(private) + elif attack == "hardlink": + os.link(private, root / ref.digest) + elif attack == "fifo": + os.mkfifo(root / ref.digest, 0o600) + else: + ref = ref.model_copy(update={"digest": "../outside"}) + with pytest.raises(ArtifactError): + store.read(ref) + finally: + store.close() + + +def test_private_root_and_bounded_temporary_cleanup(tmp_path): + root = tmp_path / "root" + root.mkdir(mode=0o755) + with pytest.raises(ArtifactError, match="private"): + ArtifactStore(root) + linked = tmp_path / "linked" + linked.symlink_to(root) + with pytest.raises(ArtifactError, match="links"): + ArtifactStore(linked) + root.chmod(0o700) + store = ArtifactStore(root) + try: + old = root / (".tmp-" + uuid4().hex) + recent = root / (".tmp-" + uuid4().hex) + for path in (old, recent): + path.write_bytes(b"temporary") + path.chmod(0o600) + os.utime(old, (0, 0)) + assert store.clean_temporary(limit=1) == 1 + assert not old.exists() and recent.exists() + with pytest.raises(ArtifactError): + store.clean_temporary(min_age_seconds=0) + finally: + store.close() + + +def test_physical_quota_serializes_parallel_publishers(tmp_path): + root = tmp_path / "root" + first = ArtifactStore(root, max_bytes=6, max_files=10) + second = ArtifactStore(root, max_bytes=6, max_files=10) + + async def run(): + results = await asyncio.gather( + asyncio.to_thread(first.put, b"abcd"), + asyncio.to_thread(second.put, b"efgh"), + return_exceptions=True, + ) + assert sum(isinstance(r, ArtifactReference) for r in results) == 1 + assert sum(isinstance(r, ArtifactError) for r in results) == 1 + assert sum(p.stat().st_size for p in root.iterdir() if p.name != ".lock") == 4 + + try: + asyncio.run(run()) + finally: + first.close() + second.close() + + +def test_real_metadata_rollback_reference_fencing_and_collection(postgres_url, tmp_path): + store = ArtifactStore(tmp_path / "bytes") + artifacts = ArtifactRepository(store) + + async def run(): + engine = create_async_engine(postgres_url) + sessions = async_sessionmaker(engine, expire_on_commit=False) + ref = store.reference(b"rolled back") + try: + with pytest.raises(RuntimeError): + async with sessions.begin() as s: + await s.execute(text("SET LOCAL ROLE kf_oj_platform")) + await artifacts.put(s, b"rolled back") + raise RuntimeError("transaction aborted") + async with sessions() as s: + assert await s.get(Artifact, ref.digest) is None + assert store.read(ref) == b"rolled back" # Durable orphan, never dangling DB reference. + os.utime(tmp_path / "bytes" / ref.digest, (0, 0)) + assert await artifacts.collect_orphans(sessions) == 1 + with pytest.raises(ArtifactError): + store.read(ref) + async with sessions.begin() as s: + await s.execute(text("SET LOCAL ROLE kf_oj_platform")) + ref = await artifacts.put(s, b"durable unreferenced") + os.utime(tmp_path / "bytes" / ref.digest, (0, 0)) + assert await artifacts.collect_orphans(sessions) == 0 + async with sessions.begin() as s: + await artifacts.mark_deleting(s, ref.digest) + with pytest.raises(ArtifactError): + async with sessions.begin() as s: + await artifacts.put(s, b"durable unreferenced") + assert await artifacts.finish_deleting(sessions) == 1 + async with sessions() as s: + assert list(await s.scalars(select(Artifact))) == [] + finally: + await engine.dispose() + + try: + asyncio.run(run()) + finally: + store.close() + + +def test_bounded_metadata_collector_preserves_references_and_resumes(postgres_url, tmp_path): + from datetime import timedelta + + from test_problems import import_fixture, platform_role + + from kf_oj.modules.problems.service import ProblemService + + root = tmp_path / "import" + _, policy = import_fixture(root) + store = ArtifactStore(tmp_path / "bytes") + repository = ArtifactRepository(store) + + async def run(): + engine = create_async_engine(postgres_url) + sessions = async_sessionmaker(engine, expire_on_commit=False) + try: + service = ProblemService(sessions, repository, policy) + version = await service.import_directory(root) + async with sessions.begin() as session: + orphan = await repository.put(session, b"committed without business reference") + await session.execute( + update(Artifact).values(created_at=func.now() - timedelta(days=2)) + ) + # After trusted fixture preparation, run collection with actual limited platform rights. + await engine.dispose() + platform_role(engine) + marked_total, cursor = 0, None + for _ in range(100): + marked, cursor = await repository.collect_metadata(sessions, limit=1, after=cursor) + marked_total += marked + if cursor is None: + break + else: + pytest.fail("bounded metadata scan failed to finish") + assert marked_total == 1 + assert await repository.finish_deleting(sessions, limit=1) == 1 + assert await repository.finish_deleting(sessions, limit=1) == 0 + async with sessions() as session: + projected = await service.for_submission(session, version) + assert ( + await repository.read(session, projected.cases[0].expected) + == b"hidden-expected\x00\xff" + ) + assert await session.get(Artifact, orphan.digest) is None + finally: + await engine.dispose() + + try: + asyncio.run(run()) + finally: + store.close() diff --git a/backend/tests/test_problems.py b/backend/tests/test_problems.py new file mode 100644 index 0000000..49485a2 --- /dev/null +++ b/backend/tests/test_problems.py @@ -0,0 +1,262 @@ +import asyncio +import base64 +import json +import os +import subprocess +import sys +from pathlib import Path +from uuid import uuid4 + +import httpx +import pytest +from sqlalchemy import event, func, select, update +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from kf_oj.api.app import create_app +from kf_oj.contracts.judging import AdmissionPolicy +from kf_oj.core.settings import Settings +from kf_oj.infra.artifacts.repository import ArtifactRepository +from kf_oj.infra.artifacts.store import ArtifactError, ArtifactStore +from kf_oj.modules.problems.models import Problem, ProblemVersion +from kf_oj.modules.problems.service import ProblemService + +ROOT = Path(__file__).resolve().parents[2] + + +def import_fixture(root, *, slug="fixture-problem"): + root.mkdir(mode=0o700) + policy = AdmissionPolicy.model_validate_json( + (ROOT / "tests/fixtures/contracts/judge-admission.valid.json").read_bytes() + ) + manifest = { + "schema_version": 1, + "slug": slug, + "title": "Synthetic problem", + "statement_file": "statement.md", + "languages": ["c17", "cpp20"], + "toolchain": policy.toolchain.model_dump(), + "comparison": "exact_bytes", + "compile_limits": policy.compile_ceiling.model_dump(), + "run_limits": policy.run_ceiling.model_dump(), + "job_wall_time_us": 60000000, + "cases": [ + {"key": "empty", "input_file": "input-1", "expected_file": "expected-1"}, + {"key": "binary", "input_file": "input-2", "expected_file": "expected-2"}, + ], + "samples": [{"input_file": "sample-input", "output_file": "sample-output"}], + } + for filename, data in { + "statement.md": b"Public statement only", + "input-1": b"", + "expected-1": b"hidden-expected\x00\xff", + "input-2": b"hidden-input\x00\xfe", + "expected-2": b"hidden-expected-2\xff", + "sample-input": b"1\n", + "sample-output": b"2\n", + }.items(): + (root / filename).write_bytes(data) + (root / "problem.json").write_text(json.dumps(manifest)) + return manifest, policy + + +def platform_role(engine): + @event.listens_for(engine.sync_engine, "connect") + def apply_role(connection, _): + cursor = connection.cursor() + cursor.execute("SET ROLE kf_oj_platform") + cursor.close() + + +def test_controlled_import_cli_and_unpublished_version_visibility(postgres_url, tmp_path): + root = tmp_path / "import" + _, policy = import_fixture(root) + env = os.environ | { + "KF_OJ_DATABASE_URL": postgres_url, + "KF_OJ_ARTIFACT_ROOT": str(tmp_path / "artifacts"), + "KF_OJ_JUDGE_ADMISSION": policy.model_dump_json(), + "KF_OJ_ENVIRONMENT": "test", + } + result = subprocess.run( + [sys.executable, "-m", "kf_oj.modules.problems.cli", str(root)], + env=env, + capture_output=True, + text=True, + timeout=15, + ) + assert result.returncode == 0 and "Problem version published:" in result.stdout + assert "hidden-expected" not in result.stdout + result.stderr + + async def run(): + engine = create_async_engine(postgres_url) + platform_role(engine) + sessions = async_sessionmaker(engine, expire_on_commit=False) + try: + service = ProblemService(sessions) + page = await service.list() + assert page.items[0].revision == 1 + version_id = page.items[0].version_id + draft_id = uuid4() + async with sessions.begin() as session: + published = await session.get(ProblemVersion, version_id) + session.add( + ProblemVersion( + id=draft_id, + problem_id=published.problem_id, + revision=2, + title=published.title, + statement=published.statement, + languages=published.languages, + toolchain=published.toolchain, + compile_limits=published.compile_limits, + run_limits=published.run_limits, + job_wall_time_us=published.job_wall_time_us, + public_samples=[], + ) + ) + from kf_oj.modules.problems.public import ProblemUnavailable + + with pytest.raises(ProblemUnavailable): + await service.detail(draft_id) + assert (await service.list()).items[0].version_id == version_id + finally: + await engine.dispose() + + asyncio.run(run()) + + +def test_actual_import_two_immutable_versions_and_public_api(postgres_url, tmp_path): + root = tmp_path / "import" + manifest, policy = import_fixture(root) + + async def run(): + app = create_app( + Settings( + _env_file=None, + environment="test", + database_url=postgres_url, + artifact_root=tmp_path / "artifacts", + judge_admission=policy, + ) + ) + async with app.router.lifespan_context(app): + platform_role(app.state.sessions.kw["bind"]) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + assert (await client.get("/api/v1/problems")).json() == { + "items": [], + "next_after": None, + } + first = await app.state.problems.import_directory(root) + async with app.state.sessions() as session: + projected = await app.state.problems.for_submission(session, first) + assert len(projected.cases) == 2 and projected.cases[0].input.bytes == 0 + binary = await app.state.problems.artifacts.read( + session, projected.cases[0].expected + ) + assert binary == b"hidden-expected\x00\xff" + manifest["title"] = "Second immutable version" + (root / "statement.md").write_text("New public statement") + (root / "problem.json").write_text(json.dumps(manifest)) + second = await app.state.problems.import_directory(root) + old = await client.get(f"/api/v1/problems/versions/{first}") + new = await client.get(f"/api/v1/problems/versions/{second}") + assert old.json()["revision"] == 1 and new.json()["revision"] == 2 + assert old.json()["statement"] == "Public statement only" + assert new.json()["statement"] == "New public statement" + assert new.json()["samples"][0] == { + "encoding": "base64", + "input": base64.b64encode(b"1\n").decode(), + "output": base64.b64encode(b"2\n").decode(), + } + page = (await client.get("/api/v1/problems")).json() + assert page["items"][0]["version_id"] == str(second) + assert (await client.get("/api/v1/problems?limit=0")).status_code == 422 + assert (await client.get(f"/api/v1/problems/versions/{uuid4()}")).status_code == 404 + public_text = old.text + new.text + json.dumps(page) + assert "hidden-expected" not in public_text and "hidden-input" not in public_text + assert str(root) not in public_text and "digest" not in public_text + async with app.state.sessions.begin() as session: + await session.execute(update(Problem).values(enabled=False)) + assert (await client.get(f"/api/v1/problems/versions/{first}")).status_code == 404 + assert (await client.get("/api/v1/problems")).json()["items"] == [] + + asyncio.run(run()) + + +@pytest.mark.parametrize("invalid", ["missing", "duplicate", "deadline", "node_budget"]) +def test_import_invalid_package_refused_without_published_rows(postgres_url, tmp_path, invalid): + root = tmp_path / "import" + manifest, policy = import_fixture(root) + if invalid == "missing": + (root / "expected-1").unlink() + elif invalid == "duplicate": + manifest["cases"].append(manifest["cases"][0]) + elif invalid == "deadline": + manifest["job_wall_time_us"] = 1000 + else: + manifest["run_limits"]["cpu_time_us"] = policy.run_ceiling.cpu_time_us + 1 + (root / "problem.json").write_text(json.dumps(manifest)) + store = ArtifactStore(tmp_path / "artifacts") + + async def run(): + engine = create_async_engine(postgres_url) + platform_role(engine) + sessions = async_sessionmaker(engine, expire_on_commit=False) + try: + service = ProblemService(sessions, ArtifactRepository(store), policy) + with pytest.raises(ValueError): + await service.import_directory(root) + async with sessions() as session: + assert await session.scalar(select(func.count()).select_from(ProblemVersion)) == 0 + finally: + await engine.dispose() + + try: + asyncio.run(run()) + finally: + store.close() + + +def test_concurrent_import_revisions_pagination_and_storage_failure_rollback( + postgres_url, tmp_path +): + root = tmp_path / "import" + _, policy = import_fixture(root) + store = ArtifactStore(tmp_path / "artifacts") + + async def run(): + engine = create_async_engine(postgres_url) + platform_role(engine) + sessions = async_sessionmaker(engine, expire_on_commit=False) + try: + service = ProblemService(sessions, ArtifactRepository(store), policy) + versions = await asyncio.gather( + service.import_directory(root), service.import_directory(root) + ) + assert versions[0] != versions[1] + assert sorted([(await service.detail(v)).revision for v in versions]) == [1, 2] + other = tmp_path / "other" + import_fixture(other, slug="another-problem") + await service.import_directory(other) + page = await service.list(limit=1) + assert page.next_after == "another-problem" and len(page.items) == 1 + next_page = await service.list(after=page.next_after, limit=1) + assert next_page.next_after is None and next_page.items[0].slug == "fixture-problem" + limited = ArtifactStore(tmp_path / "limited", max_bytes=1, max_files=5) + try: + with pytest.raises(ArtifactError): + await ProblemService( + sessions, ArtifactRepository(limited), policy + ).import_directory(root) + finally: + limited.close() + async with sessions() as session: + assert await session.scalar(select(func.count()).select_from(ProblemVersion)) == 3 + finally: + await engine.dispose() + + try: + asyncio.run(run()) + finally: + store.close() diff --git a/backend/tests/test_worker_artifacts.py b/backend/tests/test_worker_artifacts.py new file mode 100644 index 0000000..8b23846 --- /dev/null +++ b/backend/tests/test_worker_artifacts.py @@ -0,0 +1,355 @@ +import asyncio +import hashlib +import secrets +from datetime import timedelta +from pathlib import Path +from uuid import uuid4 + +import httpx +import pytest +from sqlalchemy import func, select, text, update +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from test_problems import import_fixture, platform_role + +from kf_oj.api.app import create_app +from kf_oj.contracts.judging import ( + JobManifest, + LeaseBinding, + manifest_bytes, +) +from kf_oj.core.settings import Settings +from kf_oj.modules.identity.service import token_digest +from kf_oj.modules.submissions.models import Evaluation, JudgeAttempt, Submission +from kf_oj.modules.submissions.public import SubmissionUnavailable +from kf_oj.modules.submissions.queries import SubmissionQueries + + +def test_canonical_manifest_snapshot(): + root = Path(__file__).resolve().parents[2] / "tests/fixtures/contracts" + manifest = JobManifest.model_validate_json((root / "judge-manifest.valid.json").read_bytes()) + assert manifest_bytes(manifest) == (root / "canonical-manifest.v1.txt").read_bytes() + + +async def install_attempt(sessions, app, account, version_id, *, source=b"synthetic source"): + # Explicit controlled DB fixture, not a simulated Coordinator API or claimed execution. + # E implements actual transactional claim; here the real B01 schema protects associations. + submission_id, evaluation_id, attempt_id, worker_id = (uuid4() for _ in range(4)) + lease_token = secrets.token_hex(32) + async with app.state.sessions.begin() as session: + source_ref = await app.state.problems.artifacts.put(session, source) + projected = await app.state.problems.for_submission(session, version_id) + session.add( + Submission( + id=submission_id, + account_id=account.account_id, + problem_id=projected.problem_id, + problem_version_id=version_id, + language="c17", + source_digest=source_ref.digest, + idempotency_key=uuid4(), + request_digest="e" * 64, + ) + ) + await session.flush() + # Use column-level INSERT. ORM inserts optional execution fields, which the platform + # deliberately cannot assign; application creation must use the same column projection. + await session.execute( + Evaluation.__table__.insert().values( + id=evaluation_id, submission_id=submission_id, problem_version_id=version_id + ) + ) + manifest = JobManifest( + schema_version=1, + kind="judge.manifest", + evaluation_id=evaluation_id, + submission_id=submission_id, + problem_version_id=version_id, + language="c17", + toolchain=projected.toolchain, + source=source_ref, + cases=projected.cases, + comparison="exact_bytes", + compile_limits=projected.compile_limits, + run_limits=projected.run_limits, + job_wall_time_us=projected.job_wall_time_us, + ) + async with sessions.begin() as session: + await session.execute(text("SET LOCAL ROLE kf_oj_coordinator")) + now = await session.scalar(select(func.now())) + session.add( + JudgeAttempt( + id=attempt_id, + evaluation_id=evaluation_id, + worker_id=worker_id, + number=1, + fencing_version=1, + lease_token_hash=token_digest(lease_token), + expires_at=now + timedelta(minutes=10), + manifest=manifest.model_dump(mode="json"), + manifest_digest=hashlib.sha256(manifest_bytes(manifest)).hexdigest(), + ) + ) + await session.flush() + await session.execute( + update(Evaluation) + .where(Evaluation.id == evaluation_id) + .values( + current_attempt_id=attempt_id, state="claimed", attempts_count=1, fencing_version=1 + ) + ) + return ( + LeaseBinding( + evaluation_id=evaluation_id, + attempt_id=attempt_id, + worker_id=worker_id, + fencing_version=1, + ), + lease_token, + manifest, + ) + + +def test_real_two_user_two_version_manifest_and_artifact_authorization(postgres_url, tmp_path): + root = tmp_path / "import" + _, policy = import_fixture(root) + + async def run(): + owner = create_async_engine(postgres_url) + owner_sessions = async_sessionmaker(owner, expire_on_commit=False) + app = create_app( + Settings( + _env_file=None, + environment="test", + database_url=postgres_url, + artifact_root=tmp_path / "artifacts", + judge_admission=policy, + ) + ) + try: + async with app.router.lifespan_context(app): + platform_role(app.state.sessions.kw["bind"]) + password = secrets.token_hex(16) + first_user = await app.state.identity.create_account("first_user", password) + second_password = secrets.token_hex(16) + second_user = await app.state.identity.create_account( + "second_user", second_password + ) + old_version = await app.state.problems.import_directory(root) + first, token, original = await install_attempt( + owner_sessions, app, first_user, old_version + ) + (root / "expected-1").write_bytes(b"different hidden version\x00") + new_version = await app.state.problems.import_directory(root) + second, other_token, other_manifest = await install_attempt( + owner_sessions, app, second_user, new_version, source=b"other user's source" + ) + queries = SubmissionQueries(app.state.sessions, app.state.problems.artifacts) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://localhost:8000" + ) as client: + login = await client.post( + "/api/v1/auth/login", + headers={"Origin": "http://localhost:8000"}, + json={"username": "first_user", "password": password}, + ) + assert login.status_code == 200 + authenticated = await app.state.identity.authenticate( + client.cookies.get("kf_session") + ) + assert ( + await queries.owned_source(authenticated, original.submission_id) + == b"synthetic source" + ) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://localhost:8000" + ) as other_client: + other_login = await other_client.post( + "/api/v1/auth/login", + headers={"Origin": "http://localhost:8000"}, + json={"username": "second_user", "password": second_password}, + ) + assert other_login.status_code == 200 + other_principal = await app.state.identity.authenticate( + other_client.cookies.get("kf_session") + ) + assert other_principal.account_id == second_user.account_id + with pytest.raises(SubmissionUnavailable): + await queries.owned_source(other_principal, original.submission_id) + path = ( + f"/api/v1/internal/evaluations/{first.evaluation_id}" + f"/attempts/{first.attempt_id}" + ) + headers = { + "Authorization": "Bearer " + token, + "X-Worker-ID": str(first.worker_id), + "X-Fencing-Version": "1", + } + assert ( + await client.get(path + "/manifest", headers=headers) + ).content == manifest_bytes(original) + # Changing the active problem version did not change an existing evaluation. + async with owner_sessions() as s: + evaluation = await s.get(Evaluation, first.evaluation_id) + assert evaluation.problem_version_id == old_version + assert ( + await client.get( + path + "/manifest", + headers={k: v for k, v in headers.items() if k != "Authorization"}, + ) + ).status_code == 401 + assert ( + await client.get( + path + "/manifest", + headers=headers | {"Authorization": "Bearer " + other_token}, + ) + ).status_code == 404 + assert ( + await client.get( + path + "/manifest", headers=headers | {"X-Worker-ID": str(uuid4())} + ) + ).status_code == 404 + assert ( + await client.get( + path + "/manifest", headers=headers | {"X-Fencing-Version": "2"} + ) + ).status_code == 404 + for reference in ( + original.source, + original.cases[0].input, + original.cases[0].expected, + ): + response = await client.get( + path + "/artifacts/" + reference.digest, headers=headers + ) + assert response.status_code == 200 + assert len(response.content) == reference.bytes + assert hashlib.sha256(response.content).hexdigest() == reference.digest + assert response.headers["cache-control"] == "no-store" + assert ( + await client.get( + path + "/artifacts/" + other_manifest.source.digest, headers=headers + ) + ).status_code == 404 + assert ( + await client.get( + path + "/artifacts/" + other_manifest.cases[0].expected.digest, + headers=headers, + ) + ).status_code == 404 + assert ( + await client.get(path + "/artifacts/" + "f" * 64, headers=headers) + ).status_code == 404 + second_path = ( + f"/api/v1/internal/evaluations/{second.evaluation_id}" + f"/attempts/{second.attempt_id}" + ) + assert ( + await client.get(second_path + "/manifest", headers=headers) + ).status_code == 404 + assert ( + await client.get( + path + "/manifest", + headers={ + "Authorization": "Bearer " + client.cookies.get("kf_session"), + "X-Worker-ID": str(first.worker_id), + "X-Fencing-Version": "1", + }, + ) + ).status_code == 404 + finally: + await owner.dispose() + + asyncio.run(run()) + + +@pytest.mark.parametrize( + "invalid", ["expired", "revoked", "cancel_requested", "terminal", "old_attempt"] +) +def test_expired_revoked_cancelled_terminal_or_old_attempt_rejected( + postgres_url, tmp_path, invalid +): + root = tmp_path / "import" + _, policy = import_fixture(root) + + async def run(): + owner = create_async_engine(postgres_url) + sessions = async_sessionmaker(owner, expire_on_commit=False) + app = create_app( + Settings( + _env_file=None, + environment="test", + database_url=postgres_url, + artifact_root=tmp_path / "artifacts", + judge_admission=policy, + ) + ) + try: + async with app.router.lifespan_context(app): + platform_role(app.state.sessions.kw["bind"]) + account = await app.state.identity.create_account( + "fixture_user", secrets.token_hex(16) + ) + version = await app.state.problems.import_directory(root) + binding, token, manifest = await install_attempt(sessions, app, account, version) + async with sessions.begin() as s: + await s.execute(text("SET LOCAL ROLE kf_oj_coordinator")) + if invalid == "expired": + await s.execute( + update(JudgeAttempt).values( + expires_at=func.now() - timedelta(seconds=1) + ) + ) + elif invalid == "revoked": + await s.execute(update(JudgeAttempt).values(revoked_at=func.now())) + elif invalid == "cancel_requested": + await s.execute(update(Evaluation).values(cancel_requested_at=func.now())) + elif invalid == "terminal": + await s.execute(update(Evaluation).values(state="finished")) + else: + new_attempt = uuid4() + now = await s.scalar(select(func.now())) + s.add( + JudgeAttempt( + id=new_attempt, + evaluation_id=binding.evaluation_id, + worker_id=uuid4(), + number=2, + fencing_version=2, + lease_token_hash="e" * 64, + expires_at=now + timedelta(minutes=1), + manifest=manifest.model_dump(mode="json"), + manifest_digest=hashlib.sha256( + manifest_bytes(manifest) + ).hexdigest(), + ) + ) + await s.flush() + await s.execute( + update(Evaluation).values( + current_attempt_id=new_attempt, fencing_version=2, attempts_count=2 + ) + ) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + prefix = ( + f"/api/v1/internal/evaluations/{binding.evaluation_id}" + f"/attempts/{binding.attempt_id}" + ) + headers = { + "Authorization": "Bearer " + token, + "X-Worker-ID": str(binding.worker_id), + "X-Fencing-Version": "1", + } + assert ( + await client.get(prefix + "/manifest", headers=headers) + ).status_code == 404 + assert ( + await client.get( + prefix + "/artifacts/" + manifest.source.digest, headers=headers + ) + ).status_code == 404 + finally: + await owner.dispose() + + asyncio.run(run()) diff --git a/contracts/README.md b/contracts/README.md index 0f62d94..e6074a3 100644 --- a/contracts/README.md +++ b/contracts/README.md @@ -33,3 +33,8 @@ Coordinator 必须在事务锁内核对当前租约及状态规则，并将结 `cancelled`；清理失败必须保留独立故障，缺失计量始终为 null。 生成器 `export.py` 是开发工具，允许读取 API 组合层；运行期的契约模型禁止依赖业务模块。 +# 任务清单字节身份 + +Python/Rust 的 `manifest_bytes` 输出同一 UTF-8 JSON：递归排序对象键，无额外空白/换行。 +清单摘要是这些确切字节的 SHA-256，不能使用数据库 JSONB 的输出顺序或任意 prettify 后的字节。 +共享 `canonical-manifest.v1.txt` 为合成示例的字节验收，平台内部响应复用这些序列化字节。 diff --git a/contracts/openapi/platform.v1.json b/contracts/openapi/platform.v1.json index 4040e32..b5c6b13 100644 --- a/contracts/openapi/platform.v1.json +++ b/contracts/openapi/platform.v1.json @@ -1,6 +1,30 @@ { "components": { "schemas": { + "ArtifactReference": { + "additionalProperties": false, + "properties": { + "bytes": { + "maximum": 67108864.0, + "minimum": 0.0, + "title": "Bytes", + "type": "integer" + }, + "digest": { + "maxLength": 64, + "minLength": 64, + "pattern": "^[0-9a-f]{64}$", + "title": "Digest", + "type": "string" + } + }, + "required": [ + "digest", + "bytes" + ], + "title": "ArtifactReference", + "type": "object" + }, "CurrentUser": { "additionalProperties": false, "properties": { @@ -60,6 +84,96 @@ "title": "HealthResponse", "type": "object" }, + "JobManifest": { + "additionalProperties": false, + "properties": { + "cases": { + "items": { + "$ref": "#/components/schemas/TestCase" + }, + "maxItems": 256, + "minItems": 1, + "title": "Cases", + "type": "array" + }, + "comparison": { + "const": "exact_bytes", + "title": "Comparison", + "type": "string" + }, + "compile_limits": { + "$ref": "#/components/schemas/RunLimits" + }, + "evaluation_id": { + "format": "uuid", + "title": "Evaluation Id", + "type": "string" + }, + "job_wall_time_us": { + "maximum": 3600000000.0, + "minimum": 1000.0, + "title": "Job Wall Time Us", + "type": "integer" + }, + "kind": { + "const": "judge.manifest", + "title": "Kind", + "type": "string" + }, + "language": { + "$ref": "#/components/schemas/Language" + }, + "problem_version_id": { + "format": "uuid", + "title": "Problem Version Id", + "type": "string" + }, + "run_limits": { + "$ref": "#/components/schemas/RunLimits" + }, + "schema_version": { + "const": 1, + "title": "Schema Version", + "type": "integer" + }, + "source": { + "$ref": "#/components/schemas/ArtifactReference" + }, + "submission_id": { + "format": "uuid", + "title": "Submission Id", + "type": "string" + }, + "toolchain": { + "$ref": "#/components/schemas/Toolchain" + } + }, + "required": [ + "schema_version", + "kind", + "evaluation_id", + "submission_id", + "problem_version_id", + "language", + "toolchain", + "source", + "cases", + "comparison", + "compile_limits", + "run_limits", + "job_wall_time_us" + ], + "title": "JobManifest", + "type": "object" + }, + "Language": { + "enum": [ + "c17", + "cpp20" + ], + "title": "Language", + "type": "string" + }, "LoginRequest": { "additionalProperties": false, "properties": { @@ -111,6 +225,300 @@ "title": "LoginResponse", "type": "object" }, + "ProblemDetail": { + "additionalProperties": false, + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "languages": { + "items": { + "$ref": "#/components/schemas/Language" + }, + "title": "Languages", + "type": "array" + }, + "revision": { + "title": "Revision", + "type": "integer" + }, + "run_limits": { + "$ref": "#/components/schemas/RunLimits" + }, + "samples": { + "items": { + "$ref": "#/components/schemas/PublicSample" + }, + "title": "Samples", + "type": "array" + }, + "slug": { + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$", + "title": "Slug", + "type": "string" + }, + "statement": { + "title": "Statement", + "type": "string" + }, + "title": { + "title": "Title", + "type": "string" + }, + "version_id": { + "format": "uuid", + "title": "Version Id", + "type": "string" + } + }, + "required": [ + "id", + "slug", + "version_id", + "revision", + "title", + "languages", + "statement", + "samples", + "run_limits" + ], + "title": "ProblemDetail", + "type": "object" + }, + "ProblemPage": { + "additionalProperties": false, + "properties": { + "items": { + "items": { + "$ref": "#/components/schemas/ProblemSummary" + }, + "title": "Items", + "type": "array" + }, + "next_after": { + "anyOf": [ + { + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Next After" + } + }, + "required": [ + "items", + "next_after" + ], + "title": "ProblemPage", + "type": "object" + }, + "ProblemSummary": { + "additionalProperties": false, + "properties": { + "id": { + "format": "uuid", + "title": "Id", + "type": "string" + }, + "languages": { + "items": { + "$ref": "#/components/schemas/Language" + }, + "title": "Languages", + "type": "array" + }, + "revision": { + "title": "Revision", + "type": "integer" + }, + "slug": { + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$", + "title": "Slug", + "type": "string" + }, + "title": { + "title": "Title", + "type": "string" + }, + "version_id": { + "format": "uuid", + "title": "Version Id", + "type": "string" + } + }, + "required": [ + "id", + "slug", + "version_id", + "revision", + "title", + "languages" + ], + "title": "ProblemSummary", + "type": "object" + }, + "PublicSample": { + "additionalProperties": false, + "properties": { + "encoding": { + "const": "base64", + "default": "base64", + "title": "Encoding", + "type": "string" + }, + "input": { + "maxLength": 87384, + "title": "Input", + "type": "string" + }, + "output": { + "maxLength": 87384, + "title": "Output", + "type": "string" + } + }, + "required": [ + "input", + "output" + ], + "title": "PublicSample", + "type": "object" + }, + "RunLimits": { + "additionalProperties": false, + "properties": { + "cpu_time_us": { + "maximum": 60000000.0, + "minimum": 1000.0, + "title": "Cpu Time Us", + "type": "integer" + }, + "file_bytes": { + "maximum": 67108864.0, + "minimum": 1.0, + "title": "File Bytes", + "type": "integer" + }, + "memory_bytes": { + "maximum": 2147483648.0, + "minimum": 1.0, + "title": "Memory Bytes", + "type": "integer" + }, + "open_files": { + "maximum": 1024.0, + "minimum": 3.0, + "title": "Open Files", + "type": "integer" + }, + "processes": { + "maximum": 128.0, + "minimum": 1.0, + "title": "Processes", + "type": "integer" + }, + "stack_bytes": { + "maximum": 268435456.0, + "minimum": 1.0, + "title": "Stack Bytes", + "type": "integer" + }, + "stderr_bytes": { + "maximum": 16777216.0, + "minimum": 1.0, + "title": "Stderr Bytes", + "type": "integer" + }, + "stdout_bytes": { + "maximum": 16777216.0, + "minimum": 1.0, + "title": "Stdout Bytes", + "type": "integer" + }, + "wall_time_us": { + "maximum": 120000000.0, + "minimum": 1000.0, + "title": "Wall Time Us", + "type": "integer" + }, + "workdir_bytes": { + "maximum": 268435456.0, + "minimum": 1.0, + "title": "Workdir Bytes", + "type": "integer" + } + }, + "required": [ + "cpu_time_us", + "wall_time_us", + "memory_bytes", + "stack_bytes", + "processes", + "open_files", + "stdout_bytes", + "stderr_bytes", + "file_bytes", + "workdir_bytes" + ], + "title": "RunLimits", + "type": "object" + }, + "TestCase": { + "additionalProperties": false, + "properties": { + "case_id": { + "format": "uuid", + "title": "Case Id", + "type": "string" + }, + "expected": { + "$ref": "#/components/schemas/ArtifactReference" + }, + "input": { + "$ref": "#/components/schemas/ArtifactReference" + } + }, + "required": [ + "case_id", + "input", + "expected" + ], + "title": "TestCase", + "type": "object" + }, + "Toolchain": { + "additionalProperties": false, + "properties": { + "image_kind": { + "const": "linux_gnu_cxx_x86_64_v1", + "title": "Image Kind", + "type": "string" + }, + "image_revision": { + "maximum": 4294967295.0, + "minimum": 1.0, + "title": "Image Revision", + "type": "integer" + }, + "toolchain_id": { + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$", + "title": "Toolchain Id", + "type": "string" + } + }, + "required": [ + "toolchain_id", + "image_kind", + "image_revision" + ], + "title": "Toolchain", + "type": "object" + }, "ValidationError": { "properties": { "ctx": { @@ -151,6 +559,12 @@ "title": "ValidationError", "type": "object" } + }, + "securitySchemes": { + "WorkerLease": { + "scheme": "bearer", + "type": "http" + } } }, "info": { @@ -257,6 +671,277 @@ "health" ] } + }, + "/api/v1/internal/evaluations/{evaluation_id}/attempts/{attempt_id}/artifacts/{digest}": { + "get": { + "operationId": "artifact_api_v1_internal_evaluations__evaluation_id__attempts__attempt_id__artifacts__digest__get", + "parameters": [ + { + "in": "path", + "name": "evaluation_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Evaluation Id", + "type": "string" + } + }, + { + "in": "path", + "name": "attempt_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Attempt Id", + "type": "string" + } + }, + { + "in": "path", + "name": "digest", + "required": true, + "schema": { + "title": "Digest", + "type": "string" + } + }, + { + "in": "header", + "name": "X-Worker-ID", + "required": true, + "schema": { + "format": "uuid", + "title": "X-Worker-Id", + "type": "string" + } + }, + { + "in": "header", + "name": "X-Fencing-Version", + "required": true, + "schema": { + "minimum": 1, + "title": "X-Fencing-Version", + "type": "integer" + } + } + ], + "responses": { + "200": { + "content": { + "application/octet-stream": { + "schema": { + "format": "binary", + "type": "string" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "security": [ + { + "WorkerLease": [] + } + ], + "summary": "Artifact", + "tags": [ + "worker artifacts" + ] + } + }, + "/api/v1/internal/evaluations/{evaluation_id}/attempts/{attempt_id}/manifest": { + "get": { + "operationId": "manifest_api_v1_internal_evaluations__evaluation_id__attempts__attempt_id__manifest_get", + "parameters": [ + { + "in": "path", + "name": "evaluation_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Evaluation Id", + "type": "string" + } + }, + { + "in": "path", + "name": "attempt_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Attempt Id", + "type": "string" + } + }, + { + "in": "header", + "name": "X-Worker-ID", + "required": true, + "schema": { + "format": "uuid", + "title": "X-Worker-Id", + "type": "string" + } + }, + { + "in": "header", + "name": "X-Fencing-Version", + "required": true, + "schema": { + "minimum": 1, + "title": "X-Fencing-Version", + "type": "integer" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/JobManifest" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "security": [ + { + "WorkerLease": [] + } + ], + "summary": "Manifest", + "tags": [ + "worker artifacts" + ] + } + }, + "/api/v1/problems": { + "get": { + "operationId": "list_problems_api_v1_problems_get", + "parameters": [ + { + "in": "query", + "name": "after", + "required": false, + "schema": { + "anyOf": [ + { + "pattern": "^[a-z0-9][a-z0-9_-]{0,63}$", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "After" + } + }, + { + "in": "query", + "name": "limit", + "required": false, + "schema": { + "default": 20, + "maximum": 100, + "minimum": 1, + "title": "Limit", + "type": "integer" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemPage" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "List Problems", + "tags": [ + "problems" + ] + } + }, + "/api/v1/problems/versions/{version_id}": { + "get": { + "operationId": "problem_version_api_v1_problems_versions__version_id__get", + "parameters": [ + { + "in": "path", + "name": "version_id", + "required": true, + "schema": { + "format": "uuid", + "title": "Version Id", + "type": "string" + } + } + ], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ProblemDetail" + } + } + }, + "description": "Successful Response" + }, + "422": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + }, + "description": "Validation Error" + } + }, + "summary": "Problem Version", + "tags": [ + "problems" + ] + } } } } diff --git a/frontend/src/api/generated/schema.d.ts b/frontend/src/api/generated/schema.d.ts index 2d40995..b9cf25f 100644 --- a/frontend/src/api/generated/schema.d.ts +++ b/frontend/src/api/generated/schema.d.ts @@ -75,10 +75,85 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/internal/evaluations/{evaluation_id}/attempts/{attempt_id}/artifacts/{digest}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** Artifact */ + get: operations["artifact_api_v1_internal_evaluations__evaluation_id__attempts__attempt_id__artifacts__digest__get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/v1/internal/evaluations/{evaluation_id}/attempts/{attempt_id}/manifest": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** Manifest */ + get: operations["manifest_api_v1_internal_evaluations__evaluation_id__attempts__attempt_id__manifest_get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/v1/problems": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** List Problems */ + get: operations["list_problems_api_v1_problems_get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/api/v1/problems/versions/{version_id}": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** Problem Version */ + get: operations["problem_version_api_v1_problems_versions__version_id__get"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; } export type webhooks = Record; export interface components { schemas: { + /** ArtifactReference */ + ArtifactReference: { + /** Bytes */ + bytes: number; + /** Digest */ + digest: string; + }; /** CurrentUser */ CurrentUser: { /** @@ -109,6 +184,53 @@ export interface components { /** Version */ version: string; }; + /** JobManifest */ + JobManifest: { + /** Cases */ + cases: components["schemas"]["TestCase"][]; + /** + * Comparison + * @constant + */ + comparison: "exact_bytes"; + compile_limits: components["schemas"]["RunLimits"]; + /** + * Evaluation Id + * Format: uuid + */ + evaluation_id: string; + /** Job Wall Time Us */ + job_wall_time_us: number; + /** + * Kind + * @constant + */ + kind: "judge.manifest"; + language: components["schemas"]["Language"]; + /** + * Problem Version Id + * Format: uuid + */ + problem_version_id: string; + run_limits: components["schemas"]["RunLimits"]; + /** + * Schema Version + * @constant + */ + schema_version: 1; + source: components["schemas"]["ArtifactReference"]; + /** + * Submission Id + * Format: uuid + */ + submission_id: string; + toolchain: components["schemas"]["Toolchain"]; + }; + /** + * Language + * @enum {string} + */ + Language: "c17" | "cpp20"; /** LoginRequest */ LoginRequest: { /** @@ -130,6 +252,118 @@ export interface components { expires_at: string; user: components["schemas"]["CurrentUser"]; }; + /** ProblemDetail */ + ProblemDetail: { + /** + * Id + * Format: uuid + */ + id: string; + /** Languages */ + languages: components["schemas"]["Language"][]; + /** Revision */ + revision: number; + run_limits: components["schemas"]["RunLimits"]; + /** Samples */ + samples: components["schemas"]["PublicSample"][]; + /** Slug */ + slug: string; + /** Statement */ + statement: string; + /** Title */ + title: string; + /** + * Version Id + * Format: uuid + */ + version_id: string; + }; + /** ProblemPage */ + ProblemPage: { + /** Items */ + items: components["schemas"]["ProblemSummary"][]; + /** Next After */ + next_after: string | null; + }; + /** ProblemSummary */ + ProblemSummary: { + /** + * Id + * Format: uuid + */ + id: string; + /** Languages */ + languages: components["schemas"]["Language"][]; + /** Revision */ + revision: number; + /** Slug */ + slug: string; + /** Title */ + title: string; + /** + * Version Id + * Format: uuid + */ + version_id: string; + }; + /** PublicSample */ + PublicSample: { + /** + * Encoding + * @default base64 + * @constant + */ + encoding: "base64"; + /** Input */ + input: string; + /** Output */ + output: string; + }; + /** RunLimits */ + RunLimits: { + /** Cpu Time Us */ + cpu_time_us: number; + /** File Bytes */ + file_bytes: number; + /** Memory Bytes */ + memory_bytes: number; + /** Open Files */ + open_files: number; + /** Processes */ + processes: number; + /** Stack Bytes */ + stack_bytes: number; + /** Stderr Bytes */ + stderr_bytes: number; + /** Stdout Bytes */ + stdout_bytes: number; + /** Wall Time Us */ + wall_time_us: number; + /** Workdir Bytes */ + workdir_bytes: number; + }; + /** TestCase */ + TestCase: { + /** + * Case Id + * Format: uuid + */ + case_id: string; + expected: components["schemas"]["ArtifactReference"]; + input: components["schemas"]["ArtifactReference"]; + }; + /** Toolchain */ + Toolchain: { + /** + * Image Kind + * @constant + */ + image_kind: "linux_gnu_cxx_x86_64_v1"; + /** Image Revision */ + image_revision: number; + /** Toolchain Id */ + toolchain_id: string; + }; /** ValidationError */ ValidationError: { /** Context */ @@ -243,4 +477,138 @@ export interface operations { }; }; }; + artifact_api_v1_internal_evaluations__evaluation_id__attempts__attempt_id__artifacts__digest__get: { + parameters: { + query?: never; + header: { + "X-Worker-ID": string; + "X-Fencing-Version": number; + }; + path: { + evaluation_id: string; + attempt_id: string; + digest: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/octet-stream": string; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; + manifest_api_v1_internal_evaluations__evaluation_id__attempts__attempt_id__manifest_get: { + parameters: { + query?: never; + header: { + "X-Worker-ID": string; + "X-Fencing-Version": number; + }; + path: { + evaluation_id: string; + attempt_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["JobManifest"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; + list_problems_api_v1_problems_get: { + parameters: { + query?: { + after?: string | null; + limit?: number; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ProblemPage"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; + problem_version_api_v1_problems_versions__version_id__get: { + parameters: { + query?: never; + header?: never; + path: { + version_id: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ProblemDetail"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; } diff --git a/judge/crates/contracts/src/judging.rs b/judge/crates/contracts/src/judging.rs index 5ebe8c7..386671d 100644 --- a/judge/crates/contracts/src/judging.rs +++ b/judge/crates/contracts/src/judging.rs @@ -20,6 +20,28 @@ pub fn decode_wire(payload: &[u8]) -> Result Result, &'static str> { + manifest.validate()?; + let mut value = serde_json::to_value(manifest).map_err(|_| "invalid manifest")?; + fn sorted(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(map) => { + let mut entries: Vec<_> = std::mem::take(map).into_iter().collect(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); + for (_, nested) in &mut entries { + sorted(nested); + } + *map = entries.into_iter().collect(); + } + serde_json::Value::Array(values) => values.iter_mut().for_each(sorted), + _ => {} + } + } + sorted(&mut value); + serde_json::to_vec(&value).map_err(|_| "invalid manifest") +} + fn bounded(value: u64, min: u64, max: u64) -> Validation { if (min..=max).contains(&value) { Ok(()) diff --git a/judge/crates/contracts/tests/judging_examples.rs b/judge/crates/contracts/tests/judging_examples.rs index 14d786c..ea1fc8d 100644 --- a/judge/crates/contracts/tests/judging_examples.rs +++ b/judge/crates/contracts/tests/judging_examples.rs @@ -9,6 +9,15 @@ fn fixtures() -> PathBuf { fn fixture(name: &str) -> T { decode_wire(&fs::read(fixtures().join(format!("{name}.valid.json"))).unwrap()).unwrap() } + +#[test] +fn canonical_manifest_bytes_match_python_snapshot() { + let manifest: JobManifest = fixture("judge-manifest"); + assert_eq!( + manifest_bytes(&manifest).unwrap(), + fs::read(fixtures().join("canonical-manifest.v1.txt")).unwrap() + ); +} fn roundtrip(bytes: &[u8], invalid: bool, name: &str) { let result = decode_wire::(bytes); if invalid { diff --git a/tests/fixtures/contracts/canonical-manifest.v1.txt b/tests/fixtures/contracts/canonical-manifest.v1.txt new file mode 100644 index 0000000..cc62623 --- /dev/null +++ b/tests/fixtures/contracts/canonical-manifest.v1.txt @@ -0,0 +1 @@ +{"cases":[{"case_id":"aaaaaaaa-aaaa-4aaa-8aaa-000000000004","expected":{"bytes":0,"digest":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"},"input":{"bytes":0,"digest":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}},{"case_id":"aaaaaaaa-aaaa-4aaa-8aaa-000000000005","expected":{"bytes":1,"digest":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"},"input":{"bytes":1,"digest":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}},{"case_id":"aaaaaaaa-aaaa-4aaa-8aaa-000000000006","expected":{"bytes":2,"digest":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"},"input":{"bytes":2,"digest":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"}}],"comparison":"exact_bytes","compile_limits":{"cpu_time_us":10000000,"file_bytes":67108864,"memory_bytes":536870912,"open_files":128,"processes":32,"stack_bytes":16777216,"stderr_bytes":1048576,"stdout_bytes":1048576,"wall_time_us":30000000,"workdir_bytes":268435456},"evaluation_id":"aaaaaaaa-aaaa-4aaa-8aaa-000000000001","job_wall_time_us":60000000,"kind":"judge.manifest","language":"cpp20","problem_version_id":"aaaaaaaa-aaaa-4aaa-8aaa-000000000003","run_limits":{"cpu_time_us":1000000,"file_bytes":16777216,"memory_bytes":67108864,"open_files":64,"processes":8,"stack_bytes":8388608,"stderr_bytes":1048576,"stdout_bytes":1048576,"wall_time_us":3000000,"workdir_bytes":67108864},"schema_version":1,"source":{"bytes":124,"digest":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},"submission_id":"aaaaaaaa-aaaa-4aaa-8aaa-000000000002","toolchain":{"image_kind":"linux_gnu_cxx_x86_64_v1","image_revision":1,"toolchain_id":"gnu-cxx-v1"}} \ No newline at end of file