import hashlib import hmac import json import logging import re import secrets import time from contextlib import asynccontextmanager from pathlib import Path from typing import Annotated, Literal from fastapi import FastAPI, File, Form, Header, HTTPException, Request, UploadFile from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse, Response from sqlalchemy import func, select, text from sqlalchemy.exc import SQLAlchemyError from starlette.concurrency import run_in_threadpool from .config import settings from .expiry import expiration from .db import Job, initialize, sessions, active_jobs from .storage import storage from .fonts import fonts, stylesheet from .models.api import UploadResponse, RenderStatus, ErrorEnvelope from .models.schematic import Document from .uploads import read_schematic from .job_types import job_kind FRONTEND = Path(__file__).resolve().parents[2] / 'frontend' ID = re.compile(r'^[a-f0-9]{48}$') def digest(token): return hashlib.sha256(token.encode()).hexdigest() def error(code, message, status=400): raise HTTPException(status, {'code': code, 'message': message}) def authenticated(key): expected = settings().api_key return bool(expected and key and hmac.compare_digest(expected, key)) def require_api(key): if settings().api_key and not authenticated(key): error('unauthorized', 'Provide a valid X-API-Key header', 401) @asynccontextmanager async def lifespan(app): initialize() fonts() # Validate configured font allowlist before accepting requests. yield app = FastAPI(docs_url=None, redoc_url=None, responses={code: {'model': ErrorEnvelope} for code in (401,404,409,410,413,415,422,503)}, title='Native Altium Schematic Rendering API', version='1.0.0', lifespan=lifespan, description='Upload native SchDoc, poll durable jobs, retrieve versioned geometry and SVG. Partial compatibility is explicitly reported.') # Count the complete request body before Starlette spools multipart uploads. # Content-Length alone is insufficient for chunked requests. class UploadLimit: def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope['type'] != 'http': return await self.app(scope, receive, send) file_count = 2 if scope['path'] in ('/api/compare','/api/v1/comparisons','/api/v1/pcbs/comparisons') else 1 limit = file_count * settings().max_upload_bytes + 1024 * 1024 received = 0 async def bounded_receive(): nonlocal received message = await receive() received += len(message.get('body', b'')) if received > limit: raise HTTPException(413, {'code': 'upload_too_large', 'message': 'Multipart body exceeds configured size limit'}) return message await self.app(scope, bounded_receive, send) app.add_middleware(UploadLimit) @app.middleware('http') async def security(request, call_next): # Reject unauthorized POST bodies before the multipart parser reads them. if request.method == 'POST' and request.url.path in ('/api/v1/schematics/render','/api/compare','/api/v1/comparisons','/api/v1/pcbs/render','/api/v1/pcbs/comparisons'): if settings().api_key and not authenticated(request.headers.get('x-api-key')): return JSONResponse({'success': False, 'error': {'code': 'unauthorized', 'message': 'Provide a valid X-API-Key header'}}, 401) response = await call_next(request) is_viewer = request.url.path.startswith(('/view/', '/compare/')) frame_ancestors = "'self' https://git.sadentel.net" if is_viewer else "'none'" response.headers.update({ 'X-Content-Type-Options': 'nosniff', 'Referrer-Policy': 'no-referrer', 'Content-Security-Policy': f"default-src 'none'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self' data:; font-src 'self'; base-uri 'none'; frame-ancestors {frame_ancestors}; form-action 'self'", 'Cache-Control': 'no-store', 'Permissions-Policy': 'camera=(), microphone=(), geolocation=()', }) if is_viewer and 'X-Frame-Options' in response.headers: del response.headers['X-Frame-Options'] return response # Register after security so CORS also decorates its early authentication errors. app.add_middleware( CORSMiddleware, allow_origins=['https://git.sadentel.net'], allow_methods=['GET', 'POST', 'OPTIONS'], allow_headers=['X-API-Key', 'X-Viewer-Token', 'Content-Type'], ) @app.exception_handler(HTTPException) async def http_error(request, exc): detail = exc.detail if isinstance(exc.detail, dict) else {'code': 'http_error', 'message': str(exc.detail)} return JSONResponse({'success': False, 'error': detail}, status_code=exc.status_code, headers=exc.headers) @app.exception_handler(RequestValidationError) async def validation_error(request, exc): return JSONResponse({'success': False, 'error': {'code': 'invalid_request', 'message': 'Invalid or missing request fields', 'fields': ['.'.join(map(str, e['loc'])) for e in exc.errors()]}}, status_code=422) @app.get('/healthz', include_in_schema=False) def health(): with sessions()() as db: db.execute(text('SELECT 1')) return {'status': 'ok'} def authorize(identifier, key, token, *, comparison=False, resource_kind=None): if not ID.fullmatch(identifier): error('not_found', 'Schematic not found', 404) with sessions()() as db: job = db.get(Job, identifier) expected_kind = resource_kind or ('schematic_comparison' if comparison else 'schematic_render') if job is None or job_kind(job.input_key) != expected_kind: error('not_found', 'Schematic not found', 404) # Private existence and expiry are not disclosed to unauthenticated callers. if job.access == 'private' and not (authenticated(key) or (token and hmac.compare_digest(job.token_hash, digest(token)))): error('not_found', 'Schematic not found', 404) if job.expires_at != -1 and job.expires_at <= time.time(): error('link_expired', 'This schematic link has expired', 410) return job def public_status(job): result = {'success': job.status != 'failed', 'id': job.id, 'status': job.status, 'viewer_url': settings().public_url + '/view/' + job.id, 'expires_at': job.expires_at, 'access': job.access} if job.summary: result.update(json.loads(job.summary)) if job.error_code: result['error'] = {'code': job.error_code, 'message': job.error_message} return result @app.post('/api/v1/schematics/render', status_code=202, response_model=UploadResponse, response_model_exclude_none=True, responses={202: {'description': 'Queued; poll status_url'}, 401: {'description': 'API key required'}, 413: {'description': 'Upload too large'}, 422: {'description': 'Invalid file or fields'}, 503: {'description': 'Queue or storage unavailable'}}) async def upload(file: Annotated[UploadFile, File(description='Native binary OLE or ASCII .SchDoc')], access: Annotated[Literal['private', 'unlisted'], Form()] = 'private', expires_in: Annotated[int | None, Form(ge=-1, description='-1 for no expiry, or positive seconds up to configured maximum')] = None, x_api_key: Annotated[str | None, Header()] = None): require_api(x_api_key) now = time.time() expires_at = expiration(expires_in, now) name, body = await read_schematic(file) identifier = secrets.token_hex(24) token = secrets.token_urlsafe(32) job = Job(id=identifier, filename=name, status='queued', access=access, token_hash=digest(token), created_at=now, updated_at=now, expires_at=expires_at, attempt=0, input_key=f'{identifier}/input.SchDoc') def persist(): with sessions().begin() as db: # Serialize admission checks across API replicas on PostgreSQL. if db.bind.dialect.name == 'postgresql': db.execute(text('SELECT pg_advisory_xact_lock(84721309)')) pending = db.scalar(select(func.count()).select_from(Job).where(Job.status.in_(['queued','processing']), active_jobs(now))) if pending >= settings().max_pending: error('queue_full', 'Rendering queue is full; try again later', 503) try: storage().put(job.input_key, bytes(body)) except Exception: error('storage_unavailable', 'Upload storage unavailable', 503) try: db.add(job) db.flush() except Exception: storage().delete(job.input_key) raise await run_in_threadpool(persist) result = public_status(job) if access == 'private': # Fragments are never sent to servers/referrers; viewer exchanges this only in a header. result['viewer_url'] += '#token=' + token result['viewer_token'] = token result['status_url'] = settings().public_url + f'/api/v1/schematics/{identifier}/status' return JSONResponse(result, status_code=202, headers={'Location': result['status_url'], 'Retry-After': '1'}) @app.get('/api/v1/schematics/{identifier}/status', response_model=RenderStatus, response_model_exclude_none=True) def status(identifier: str, x_api_key: Annotated[str | None, Header()] = None, x_viewer_token: Annotated[str | None, Header()] = None): return public_status(authorize(identifier, x_api_key, x_viewer_token)) def artifact(identifier, kind, key, token): job = authorize(identifier, key, token) if job.status == 'failed': error(job.error_code, job.error_message, 422) if job.status != 'completed': error('not_ready', 'Rendering is still in progress', 409) try: data = storage().get(job.output_prefix + '/' + kind) except Exception: error('storage_unavailable', 'Rendering artifact unavailable', 503) return Response(data, media_type='application/json' if kind.endswith('.json') else 'image/svg+xml') @app.get('/api/v1/schematics/{identifier}/data', response_model=Document) def data(identifier: str, x_api_key: Annotated[str | None, Header()] = None, x_viewer_token: Annotated[str | None, Header()] = None): return artifact(identifier, 'data.json', x_api_key, x_viewer_token) @app.get('/api/v1/schematics/{identifier}/svg', response_class=Response, responses={200:{'content':{'image/svg+xml':{'schema':{'type':'string'}}}}}) def svg(identifier: str, x_api_key: Annotated[str | None, Header()] = None, x_viewer_token: Annotated[str | None, Header()] = None): return artifact(identifier, 'drawing.svg', x_api_key, x_viewer_token) @app.get('/view/{identifier}', include_in_schema=False) def viewer(identifier: str): if not ID.fullmatch(identifier): error('not_found', 'Schematic not found', 404) return FileResponse(FRONTEND / 'viewer.html') @app.get('/', include_in_schema=False) def index(): return FileResponse(FRONTEND / 'index.html') @app.get('/assets/{name}', include_in_schema=False) def asset(name: str): if name not in {'viewer.js','viewer.css','upload.js','upload.css','camera.js','compare.js','compare.css','compare-upload.js','interactions.js','pcb.js','pcb.css','pcb-layers.js','share.js','share.css'}: error('not_found', 'Asset not found', 404) return FileResponse(FRONTEND / name) @app.get('/docs', include_in_schema=False) def api_docs(): return FileResponse(FRONTEND / 'docs.html') @app.get('/fonts.css', include_in_schema=False) def font_css(): return Response(stylesheet(),media_type='text/css') @app.get('/fonts/{name}', include_in_schema=False) def font_file(name: str): font=fonts().get(name) if font is None: error('not_found','Font not found',404) return FileResponse(font['file']) @app.exception_handler(SQLAlchemyError) async def database_error(request, exc): logging.getLogger(__name__).error('Database request failed', exc_info=exc) return JSONResponse({'success':False,'error':{'code':'database_unavailable','message':'Database service unavailable'}},status_code=503,headers={'Cache-Control':'no-store'}) # Comparison jobs share the existing queue, access policy and error envelope. from .compare_api import register_comparison_routes register_comparison_routes(app, authorize=authorize, require_api=require_api, error=error) from .pcb_api import register_pcb_routes register_pcb_routes(app, authorize=authorize, require_api=require_api, error=error)