kitos
394d5d9056
Enable ANN rules in ruff.toml (flake8-annotations) and resolve all 221 violations: ANN201/ANN202 — return types on 168 public/private functions: - All 28 FastAPI routers: endpoints annotated with dict/list/specific schema/ StreamingResponse/FileResponse/JSONResponse as appropriate - main.py: lifespan→AsyncGenerator[None,None], exception handlers→JSONResponse - database.py: get_db→Generator[Session,None,None], proxy methods→correct types - middleware/request_context.py: dispatch→Response with Callable call_next type ANN001/ANN002/ANN003 — 32 missing argument types: - seed_demo.py: all db parameters typed as Session - domain/unit_of_work.py: __aexit__ exc_type/exc_val/exc_tb typed with TracebackType - services: audit_service user_id→UUID|None, heatmap_service query/model/builder, notification_service test→Test, tempo_service test→Test/user→User, test_workflow_service test_id→UUID, campaign_crud **fields→object, test_crud **fields→object (4 sites) ANN401 — 16 Any usages resolved: - Domain entities (campaign/technique/threat_actor/test_entity): replaced Any with actual ORM types via TYPE_CHECKING guards to avoid circular imports - detection_rule_service: test_id/detection_rule_id/evaluator_id→UUID - score_cache: kept Any with # noqa: ANN401 (genuinely generic cache) - jira_service/tempo_service: kept Any with # noqa: ANN401 (lazy optional deps) - d3fend_import_service: _to_str(v: Any) kept with # noqa: ANN401 ANN204/ANN205/ANN206 — special/static/class methods: - database.py proxy __call__/__getattr__: *args: object/**kwargs: object - schemas/test.py model_validate: obj→object, **kwargs→object - sa_technique_repository._int_type→type All 439 unit tests pass. ruff check app/ → All checks passed! Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""Professional report generation endpoints — PDF, DOCX, HTML output."""
|
|
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request
|
|
from fastapi.responses import FileResponse
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.dependencies.auth import get_current_user, require_any_role
|
|
from app.limiter import limiter
|
|
from app.models.user import User
|
|
from app.services import report_generation_service
|
|
|
|
router = APIRouter(prefix="/reports/generate", tags=["professional-reports"])
|
|
|
|
_MEDIA_TYPES = {
|
|
"pdf": "application/pdf",
|
|
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
"html": "text/html",
|
|
}
|
|
|
|
|
|
@router.get("/purple-campaign/{campaign_id}")
|
|
@limiter.limit("5/minute")
|
|
def generate_purple_report(
|
|
request: Request,
|
|
campaign_id: UUID,
|
|
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
|
|
db: Session = Depends(get_db),
|
|
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
|
|
) -> FileResponse:
|
|
"""Generate a Purple Team campaign assessment report."""
|
|
filepath = report_generation_service.generate_purple_campaign_report(
|
|
db, str(campaign_id), output_format=format,
|
|
)
|
|
return FileResponse(
|
|
filepath,
|
|
media_type=_MEDIA_TYPES[format],
|
|
filename=f"purple_report.{format}",
|
|
)
|
|
|
|
|
|
@router.get("/coverage-summary")
|
|
@limiter.limit("5/minute")
|
|
def generate_coverage_report(
|
|
request: Request,
|
|
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
|
|
db: Session = Depends(get_db),
|
|
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
|
|
) -> FileResponse:
|
|
"""Generate an organization-wide MITRE ATT&CK coverage report."""
|
|
filepath = report_generation_service.generate_coverage_report(
|
|
db, output_format=format,
|
|
)
|
|
return FileResponse(
|
|
filepath,
|
|
media_type=_MEDIA_TYPES[format],
|
|
filename=f"coverage_report.{format}",
|
|
)
|
|
|
|
|
|
@router.get("/executive-summary")
|
|
@limiter.limit("5/minute")
|
|
def generate_executive_report(
|
|
request: Request,
|
|
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
|
|
db: Session = Depends(get_db),
|
|
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
|
|
) -> FileResponse:
|
|
"""Generate an executive security summary report."""
|
|
filepath = report_generation_service.generate_executive_summary(
|
|
db, output_format=format,
|
|
)
|
|
return FileResponse(
|
|
filepath,
|
|
media_type=_MEDIA_TYPES[format],
|
|
filename=f"executive_summary.{format}",
|
|
)
|
|
|
|
|
|
@router.get("/quarterly-summary")
|
|
@limiter.limit("5/minute")
|
|
def generate_quarterly_report(
|
|
request: Request,
|
|
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
|
|
db: Session = Depends(get_db),
|
|
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
|
|
) -> FileResponse:
|
|
"""Generate a quarterly security summary report."""
|
|
filepath = report_generation_service.generate_quarterly_summary(
|
|
db, output_format=format,
|
|
)
|
|
return FileResponse(
|
|
filepath,
|
|
media_type=_MEDIA_TYPES[format],
|
|
filename=f"quarterly_summary.{format}",
|
|
)
|
|
|
|
|
|
@router.get("/technique/{technique_id}")
|
|
@limiter.limit("5/minute")
|
|
def generate_technique_report(
|
|
request: Request,
|
|
technique_id: UUID,
|
|
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
|
|
db: Session = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
) -> FileResponse:
|
|
"""Generate a detailed report for one MITRE technique."""
|
|
filepath = report_generation_service.generate_technique_detail_report(
|
|
db, str(technique_id), output_format=format,
|
|
)
|
|
return FileResponse(
|
|
filepath,
|
|
media_type=_MEDIA_TYPES[format],
|
|
filename=f"technique_{technique_id}.{format}",
|
|
)
|