Coverage for extra_codeowners/app.py: 88%
189 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-13 09:46 +0000
1"""FastAPI application factory and lifecycle."""
3from __future__ import annotations
5import asyncio
6import os
7import socket
8import uuid
9from collections.abc import AsyncIterator
10from contextlib import asynccontextmanager
12import structlog
13from fastapi import FastAPI, HTTPException, Query, Request, Response, status
14from fastapi.responses import HTMLResponse, JSONResponse
15from prometheus_client import CONTENT_TYPE_LATEST, generate_latest
17from extra_codeowners import __version__
18from extra_codeowners.database import JobRequest, QueueStore
19from extra_codeowners.github import GitHubClient
20from extra_codeowners.logging import configure_logging
21from extra_codeowners.manifest import ManifestError, ManifestService
22from extra_codeowners.metrics import INSECURE_MODE, WEBHOOK_FAILURES, WEBHOOKS
23from extra_codeowners.service import EvaluationService, Reconciler, Worker
24from extra_codeowners.settings import Settings, get_settings
25from extra_codeowners.webhooks import (
26 MAX_WEBHOOK_BYTES,
27 WebhookError,
28 evaluation_job,
29 verify_webhook,
30)
32log = structlog.get_logger()
33NO_STORE_HEADERS = {
34 "Cache-Control": "no-store, max-age=0",
35 "Pragma": "no-cache",
36 "Referrer-Policy": "no-referrer",
37 "X-Content-Type-Options": "nosniff",
38 "Content-Security-Policy": (
39 "default-src 'none'; style-src 'unsafe-inline'; "
40 "form-action https://github.com; base-uri 'none'; frame-ancestors 'none'"
41 ),
42}
45def create_app(
46 settings: Settings | None = None,
47 *,
48 github: GitHubClient | None = None,
49 store: QueueStore | None = None,
50) -> FastAPI:
51 """Build an independently testable service instance."""
52 runtime = settings or get_settings()
53 configure_logging(runtime.log_level, json_logs=runtime.environment == "production")
55 @asynccontextmanager
56 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
57 runtime.validate_for_service()
58 owned_store = store is None
59 queue_store = store or QueueStore(runtime.database_url.get_secret_value())
60 await asyncio.to_thread(queue_store.initialize)
61 app.state.store = queue_store
63 owned_github = github is None
64 github_client = github
65 if github_client is None and runtime.github_ready: 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true
66 assert runtime.github_app_id is not None
67 assert runtime.private_key_value is not None
68 github_client = GitHubClient(
69 runtime.github_app_id,
70 runtime.private_key_value,
71 api_url=str(runtime.github_api_url),
72 api_version=runtime.github_api_version,
73 )
74 app.state.github = github_client
75 app.state.stop = asyncio.Event()
76 app.state.tasks = []
77 app.state.worker_task = None
78 app.state.reconciler_task = None
79 app.state.evaluator = None
80 INSECURE_MODE.set(int(runtime.allow_insecure_changes))
81 if runtime.allow_insecure_changes: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 log.warning(
83 "insecure_changes_enabled",
84 warning=(
85 "built-in non-delegable paths are disabled; organization guardrails remain"
86 ),
87 )
89 manifest_service: ManifestService | None = None
90 if runtime.setup_enabled:
91 manifest_service = ManifestService(runtime)
92 app.state.manifest = manifest_service
94 if github_client is not None: 94 ↛ 113line 94 didn't jump to line 113 because the condition on line 94 was always true
95 owner = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:12]}"
96 evaluator = EvaluationService(runtime, github_client, queue_store)
97 app.state.evaluator = evaluator
98 if runtime.worker_enabled:
99 worker = Worker(runtime, queue_store, evaluator, owner)
100 worker_task = asyncio.create_task(
101 worker.run(app.state.stop), name="evaluation-worker"
102 )
103 app.state.worker_task = worker_task
104 app.state.tasks.append(worker_task)
105 if runtime.reconcile_enabled:
106 reconciler = Reconciler(runtime, github_client, queue_store, owner)
107 reconciler_task = asyncio.create_task(
108 reconciler.run(app.state.stop), name="open-pr-reconciler"
109 )
110 app.state.reconciler_task = reconciler_task
111 app.state.tasks.append(reconciler_task)
113 try:
114 yield
115 finally:
116 app.state.stop.set()
117 if app.state.tasks:
118 await asyncio.gather(*app.state.tasks, return_exceptions=True)
119 if manifest_service is not None:
120 await manifest_service.close()
121 if owned_github and github_client is not None: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 await github_client.close()
123 if owned_store: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 await asyncio.to_thread(queue_store.close)
126 app = FastAPI(
127 title="Extra CODEOWNERS",
128 summary="Human or delegated-application CODEOWNER approval checks",
129 version=__version__,
130 lifespan=lifespan,
131 docs_url="/api/docs",
132 redoc_url=None,
133 openapi_url="/api/openapi.json",
134 )
135 app.state.settings = runtime
137 @app.get("/", include_in_schema=False)
138 async def index() -> dict[str, str]:
139 return {
140 "name": "Extra CODEOWNERS",
141 "version": __version__,
142 "documentation": "https://extra-codeowners.readthedocs.io/",
143 }
145 @app.get("/health/live", tags=["operations"])
146 async def live(request: Request) -> JSONResponse:
147 """Return liveness, including critical in-process background tasks."""
148 worker_task: asyncio.Task[None] | None = request.app.state.worker_task
149 worker_expected = runtime.worker_enabled and request.app.state.github is not None
150 worker_alive = not worker_expected or (worker_task is not None and not worker_task.done())
151 reconciler_task: asyncio.Task[None] | None = request.app.state.reconciler_task
152 reconciler_expected = runtime.reconcile_enabled and request.app.state.github is not None
153 reconciler_alive = not reconciler_expected or (
154 reconciler_task is not None and not reconciler_task.done()
155 )
156 alive = worker_alive and reconciler_alive
157 return JSONResponse(
158 {
159 "status": "alive" if alive else "not_alive",
160 "worker": worker_alive,
161 "reconciler": reconciler_alive,
162 },
163 status_code=status.HTTP_200_OK if alive else status.HTTP_503_SERVICE_UNAVAILABLE,
164 )
166 @app.get("/health/ready", tags=["operations"])
167 async def ready(request: Request) -> JSONResponse:
168 """Return readiness only when credentials, database, and worker are usable."""
169 queue_store: QueueStore = request.app.state.store
170 database_ready = await asyncio.to_thread(queue_store.database_available)
171 worker_task: asyncio.Task[None] | None = request.app.state.worker_task
172 worker_ready = not runtime.worker_enabled or (
173 worker_task is not None and not worker_task.done()
174 )
175 reconciler_task: asyncio.Task[None] | None = request.app.state.reconciler_task
176 reconciler_ready = not runtime.reconcile_enabled or (
177 reconciler_task is not None and not reconciler_task.done()
178 )
179 ready_state = bool(
180 runtime.github_ready and database_ready and worker_ready and reconciler_ready
181 )
182 payload = {
183 "status": "ready" if ready_state else "not_ready",
184 "github_credentials": runtime.github_ready,
185 "database": database_ready,
186 "worker": worker_ready,
187 "reconciler": reconciler_ready,
188 }
189 return JSONResponse(
190 payload,
191 status_code=status.HTTP_200_OK if ready_state else status.HTTP_503_SERVICE_UNAVAILABLE,
192 )
194 @app.get("/metrics", include_in_schema=False)
195 async def metrics() -> Response:
196 """Expose Prometheus metrics without repository or secret labels."""
197 return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
199 @app.post(
200 "/webhooks/github",
201 status_code=status.HTTP_202_ACCEPTED,
202 tags=["github"],
203 responses={
204 400: {"description": "Malformed delivery"},
205 401: {"description": "Invalid signature"},
206 503: {
207 "description": (
208 "Receiver/evaluator unavailable, or delivery could not be durably stored"
209 )
210 },
211 },
212 )
213 async def github_webhook(request: Request) -> JSONResponse:
214 """Authenticate, de-duplicate, and durably enqueue a GitHub delivery."""
215 secret = runtime.webhook_secret_value
216 if secret is None: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 raise HTTPException(
218 status.HTTP_503_SERVICE_UNAVAILABLE, "webhook receiver is not configured"
219 )
220 content_length = request.headers.get("content-length")
221 if content_length is not None:
222 try:
223 parsed_length = int(content_length)
224 if parsed_length < 0: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 raise ValueError
226 if parsed_length > MAX_WEBHOOK_BYTES: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 raise HTTPException(status.HTTP_413_CONTENT_TOO_LARGE, "payload too large")
228 except ValueError as error:
229 raise HTTPException(
230 status.HTTP_400_BAD_REQUEST, "invalid Content-Length"
231 ) from error
232 body_buffer = bytearray()
233 async for chunk in request.stream():
234 body_buffer.extend(chunk)
235 if len(body_buffer) > MAX_WEBHOOK_BYTES:
236 raise HTTPException(status.HTTP_413_CONTENT_TOO_LARGE, "payload too large")
237 body = bytes(body_buffer)
238 try:
239 webhook = verify_webhook(
240 body,
241 signature=request.headers.get("x-hub-signature-256"),
242 delivery_id=request.headers.get("x-github-delivery"),
243 event=request.headers.get("x-github-event"),
244 secret=secret,
245 )
246 job = evaluation_job(
247 webhook,
248 policy_path=runtime.policy_path,
249 org_config_repository=runtime.org_config_repository,
250 )
251 if isinstance(job, JobRequest) and runtime.is_organization_config_repository(
252 job.repository_full_name
253 ):
254 # The shared organization policy occupies the repository
255 # policy path in ORG/.github, so that repository is protected
256 # by native human CODEOWNERS rules instead of this check.
257 job = None
258 except WebhookError as error:
259 reason = "signature" if "signature" in str(error).lower() else "payload"
260 WEBHOOK_FAILURES.labels(reason).inc()
261 status_code = (
262 status.HTTP_401_UNAUTHORIZED
263 if reason == "signature"
264 else status.HTTP_400_BAD_REQUEST
265 )
266 raise HTTPException(status_code, str(error)) from error
267 if job is None:
268 # Ignored actions (including this App's own check_run updates) do
269 # not need durable de-duplication. Persisting them would amplify
270 # every evaluation into retained database traffic.
271 log.info(
272 "webhook_ignored",
273 delivery_id=webhook.delivery_id,
274 github_event=webhook.event,
275 action=webhook.action or "received",
276 )
277 WEBHOOKS.labels(webhook.event, webhook.action or "received").inc()
278 return JSONResponse(
279 {"accepted": True, "queued": False},
280 status_code=status.HTTP_202_ACCEPTED,
281 )
282 queue_store: QueueStore = request.app.state.store
283 try:
284 accepted = await asyncio.to_thread(
285 queue_store.accept_delivery,
286 webhook.delivery_id,
287 webhook.event,
288 job,
289 runtime.webhook_invalidation_timeout_seconds,
290 )
291 except Exception as error:
292 WEBHOOK_FAILURES.labels("durable_acceptance").inc()
293 log.exception(
294 "webhook_durable_acceptance_failed",
295 delivery_id=webhook.delivery_id,
296 github_event=webhook.event,
297 error_type=type(error).__name__,
298 )
299 raise HTTPException(
300 status.HTTP_503_SERVICE_UNAVAILABLE,
301 "delivery could not be durably accepted; redeliver it after recovery",
302 ) from error
303 queued = accepted and job is not None
304 if isinstance(job, JobRequest) and await asyncio.to_thread(
305 queue_store.delivery_needs_invalidation, webhook.delivery_id
306 ):
307 evaluator: EvaluationService | None = request.app.state.evaluator
308 if evaluator is None: 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true
309 WEBHOOK_FAILURES.labels("invalidation_unavailable").inc()
310 raise HTTPException(
311 status.HTTP_503_SERVICE_UNAVAILABLE,
312 "delivery was stored, but check revocation is temporarily unavailable",
313 )
314 try:
315 invalidated = await asyncio.wait_for(
316 evaluator.invalidate_for_trigger(job),
317 timeout=runtime.webhook_invalidation_timeout_seconds,
318 )
319 if invalidated: 319 ↛ 326line 319 didn't jump to line 326 because the condition on line 319 was always true
320 # Always create a newer generation after the PATCH. This
321 # fences a worker that published immediately before the
322 # revocation and prevents it from deleting the only queued
323 # re-evaluation as it completes.
324 await asyncio.to_thread(queue_store.enqueue, job)
325 queued = True
326 await asyncio.to_thread(queue_store.mark_delivery_invalidated, webhook.delivery_id)
327 except Exception as error:
328 # Durable work is authoritative. GitHub terminates webhook
329 # requests after ten seconds and does not automatically retry
330 # failures, so the synchronous revocation is a bounded fast
331 # path rather than part of delivery acceptance.
332 WEBHOOK_FAILURES.labels("invalidation_fast_path").inc()
333 log.warning(
334 "webhook_check_invalidation_deferred",
335 delivery_id=webhook.delivery_id,
336 github_event=webhook.event,
337 repository=job.repository_full_name,
338 pull_number=job.pull_number,
339 error_type=type(error).__name__,
340 )
341 log.info(
342 "webhook_accepted",
343 delivery_id=webhook.delivery_id,
344 github_event=webhook.event,
345 action=webhook.action or "received",
346 repository=job.repository_full_name,
347 pull_number=job.pull_number if isinstance(job, JobRequest) else None,
348 accepted=accepted,
349 queued=queued,
350 )
351 WEBHOOKS.labels(webhook.event, webhook.action or "received").inc()
352 return JSONResponse(
353 {"accepted": accepted, "queued": queued},
354 status_code=status.HTTP_202_ACCEPTED,
355 )
357 @app.get("/setup", response_class=HTMLResponse, include_in_schema=False)
358 async def setup(
359 request: Request, organization: str | None = Query(default=None)
360 ) -> HTMLResponse:
361 """Start the operator-controlled GitHub App Manifest flow."""
362 service: ManifestService | None = request.app.state.manifest
363 if service is None:
364 raise HTTPException(status.HTTP_404_NOT_FOUND)
365 return HTMLResponse(service.registration_page(organization), headers=NO_STORE_HEADERS)
367 @app.get("/setup/callback", response_class=HTMLResponse, include_in_schema=False)
368 async def setup_callback(request: Request, code: str, state: str) -> HTMLResponse:
369 """Exchange a one-use manifest code and display credentials exactly once."""
370 service: ManifestService | None = request.app.state.manifest
371 if service is None: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 raise HTTPException(status.HTTP_404_NOT_FOUND)
373 try:
374 credentials = await service.exchange(code, state)
375 except ManifestError as error:
376 raise HTTPException(status.HTTP_400_BAD_REQUEST, str(error)) from error
377 return HTMLResponse(service.credentials_page(credentials), headers=NO_STORE_HEADERS)
379 @app.get("/setup/complete", response_class=HTMLResponse, include_in_schema=False)
380 async def setup_complete() -> HTMLResponse:
381 """Confirm installation and point the operator to repository configuration."""
382 if not runtime.setup_enabled:
383 raise HTTPException(status.HTTP_404_NOT_FOUND)
384 body = """<!doctype html><html lang="en"><head><meta charset="utf-8">
385<meta name="viewport" content="width=device-width"><title>Extra CODEOWNERS installed</title></head>
386<body><main><h1>Extra CODEOWNERS is installed</h1>
387<p>Add <code>.github/extra-codeowners.toml</code> to each repository you want to enable.</p>
388<p><a href="https://extra-codeowners.readthedocs.io/">Read the configuration guide</a>.</p>
389</main></body></html>"""
390 return HTMLResponse(body, headers=NO_STORE_HEADERS)
392 return app
395app = create_app()