Spaces:
Sleeping
Sleeping
File size: 1,618 Bytes
5f77ac0 b6ba216 5f77ac0 c05edc5 5f77ac0 c05edc5 5f77ac0 b6ba216 5f77ac0 b6ba216 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | import logging
import uvicorn
from fastapi import FastAPI, Query
from fastapi.responses import HTMLResponse, JSONResponse
from scanner.config import Settings
from scanner.pipeline import PipelineOrchestrator
from scanner.proxy import ContentSafetyProxy
logger = logging.getLogger(__name__)
def create_app(mode: str = "strip") -> FastAPI:
settings = Settings()
orchestrator = PipelineOrchestrator(settings=settings)
proxy = ContentSafetyProxy(orchestrator=orchestrator, settings=settings, mode=mode) # type: ignore[arg-type]
app = FastAPI(title="Content Safety Proxy")
@app.get("/proxy")
async def proxy_get(url: str = Query(...), mode: str = Query(None)):
"""Fetch and filter a URL."""
content, content_type, scan = await proxy.handle(url, mode=mode)
return HTMLResponse(
content=content,
headers={
"X-Scan-Status": "scanned",
"X-Risk-Score": str(scan.risk_score),
"X-Findings-Count": str(len(scan.findings)),
},
)
@app.get("/proxy/audit")
async def proxy_audit(url: str = Query(...)):
"""Get the scan report for a URL without filtering."""
report = await orchestrator.scan_url(url)
return JSONResponse(report.model_dump(mode="json"))
@app.get("/proxy/status")
async def proxy_status():
return {"status": "running", "mode": mode}
return app
def run_proxy(port: int = 9090, host: str = "127.0.0.1", mode: str = "strip"):
app = create_app(mode=mode)
uvicorn.run(app, host=host, port=port, log_level="info")
|