Files
Aegis/backend/app/routers/professional_reports.py
T
kitos dcd4bebc92
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
fix(security): resolve Snyk Code findings — Tar Slip, Path Traversal, Open Redirect, XSS
Tar Slip (CWE-22) — 3 import services:
  threat_actor, lolbas, caldera: add path validation before extractall()
  to prevent malicious zip members with ../ escaping the target directory.
  (sigma, elastic, atomic already had this protection)

Path Traversal (CWE-23) — professional_reports.py:
  Add _assert_safe_report_path() check on all 5 report endpoints to
  verify the generated filepath stays within REPORT_OUTPUT_DIR.

Open Redirect (CWE-601) — sso.py:
  Validate IdP redirect URL scheme (must be http/https) before
  issuing RedirectResponse, blocking javascript: and data: redirects.

DOM XSS (CWE-79) — 4 frontend pages:
  Create src/utils/url.ts with safeUrl() that rejects non-http/https
  protocols; apply to actor.mitre_url, ref.url, intel.url.
  Sanitize framework name to alphanumeric-only before DOM insertion.
  Restrict evidence MIME types to an explicit safe allowlist (png/jpg/gif/webp).

Hardcoded credentials (CWE-798):
  verify_gaps.py, create_wiki.py: replace literal passwords with
  environment variable reads (AEGIS_ADMIN_PASSWORD, GITEA_PASSWORD).
2026-06-12 13:15:36 +02:00

209 lines
6.8 KiB
Python

"""Professional report generation endpoints — PDF, DOCX, HTML output."""
# Import UUID from uuid
from uuid import UUID
from pathlib import Path
# Import APIRouter, Depends, HTTPException, Query, Request from fastapi
from fastapi import APIRouter, Depends, HTTPException, Query, Request
# Import FileResponse from fastapi.responses
from fastapi.responses import FileResponse
# Import Session from sqlalchemy.orm
from sqlalchemy.orm import Session
# Import get_db from app.database
from app.database import get_db
# Import get_current_user, require_any_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role
# Import limiter from app.limiter
from app.limiter import limiter
# Import settings from app.config
from app.config import settings
# Import User from app.models.user
from app.models.user import User
# Import report_generation_service from app.services
from app.services import report_generation_service
def _assert_safe_report_path(filepath: str) -> str:
"""Raise 500 if the generated filepath escapes the configured report directory."""
output_dir = Path(settings.REPORT_OUTPUT_DIR).resolve()
resolved = Path(filepath).resolve()
if not resolved.is_relative_to(output_dir):
raise HTTPException(status_code=500, detail="Report generation path error")
return filepath
# Assign router = APIRouter(prefix="/reports/generate", tags=["professional-reports"])
router = APIRouter(prefix="/reports/generate", tags=["professional-reports"])
# Assign _MEDIA_TYPES = {
_MEDIA_TYPES = {
# Literal argument value
"pdf": "application/pdf",
# Literal argument value
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
# Literal argument value
"html": "text/html",
}
# Apply the @router.get decorator
@router.get("/purple-campaign/{campaign_id}")
# Apply the @limiter.limit decorator
@limiter.limit("5/minute")
# Define function generate_purple_report
def generate_purple_report(
# Entry: request
request: Request,
# Entry: campaign_id
campaign_id: UUID,
# Entry: format
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
# Entry: db
db: Session = Depends(get_db),
# Entry: user
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
) -> FileResponse:
"""Generate a Purple Team campaign assessment report."""
# Assign filepath = report_generation_service.generate_purple_campaign_report(
filepath = report_generation_service.generate_purple_campaign_report(
db, str(campaign_id), output_format=format,
)
# Return FileResponse(
return FileResponse(
_assert_safe_report_path(filepath),
# Keyword argument: media_type
media_type=_MEDIA_TYPES[format],
# Keyword argument: filename
filename=f"purple_report.{format}",
)
# Apply the @router.get decorator
@router.get("/coverage-summary")
# Apply the @limiter.limit decorator
@limiter.limit("5/minute")
# Define function generate_coverage_report
def generate_coverage_report(
# Entry: request
request: Request,
# Entry: format
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
# Entry: db
db: Session = Depends(get_db),
# Entry: user
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
) -> FileResponse:
"""Generate an organization-wide MITRE ATT&CK coverage report."""
# Assign filepath = report_generation_service.generate_coverage_report(
filepath = report_generation_service.generate_coverage_report(
db, output_format=format,
)
# Return FileResponse(
return FileResponse(
_assert_safe_report_path(filepath),
# Keyword argument: media_type
media_type=_MEDIA_TYPES[format],
# Keyword argument: filename
filename=f"coverage_report.{format}",
)
# Apply the @router.get decorator
@router.get("/executive-summary")
# Apply the @limiter.limit decorator
@limiter.limit("5/minute")
# Define function generate_executive_report
def generate_executive_report(
# Entry: request
request: Request,
# Entry: format
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
# Entry: db
db: Session = Depends(get_db),
# Entry: user
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
) -> FileResponse:
"""Generate an executive security summary report."""
# Assign filepath = report_generation_service.generate_executive_summary(
filepath = report_generation_service.generate_executive_summary(
db, output_format=format,
)
# Return FileResponse(
return FileResponse(
_assert_safe_report_path(filepath),
# Keyword argument: media_type
media_type=_MEDIA_TYPES[format],
# Keyword argument: filename
filename=f"executive_summary.{format}",
)
# Apply the @router.get decorator
@router.get("/quarterly-summary")
# Apply the @limiter.limit decorator
@limiter.limit("5/minute")
# Define function generate_quarterly_report
def generate_quarterly_report(
# Entry: request
request: Request,
# Entry: format
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
# Entry: db
db: Session = Depends(get_db),
# Entry: user
user: User = Depends(require_any_role("red_lead", "blue_lead", "viewer")),
) -> FileResponse:
"""Generate a quarterly security summary report."""
# Assign filepath = report_generation_service.generate_quarterly_summary(
filepath = report_generation_service.generate_quarterly_summary(
db, output_format=format,
)
# Return FileResponse(
return FileResponse(
_assert_safe_report_path(filepath),
# Keyword argument: media_type
media_type=_MEDIA_TYPES[format],
# Keyword argument: filename
filename=f"quarterly_summary.{format}",
)
# Apply the @router.get decorator
@router.get("/technique/{technique_id}")
# Apply the @limiter.limit decorator
@limiter.limit("5/minute")
# Define function generate_technique_report
def generate_technique_report(
# Entry: request
request: Request,
# Entry: technique_id
technique_id: UUID,
# Entry: format
format: str = Query("pdf", pattern="^(pdf|docx|html)$"),
# Entry: db
db: Session = Depends(get_db),
# Entry: user
user: User = Depends(get_current_user),
) -> FileResponse:
"""Generate a detailed report for one MITRE technique."""
# Assign filepath = report_generation_service.generate_technique_detail_report(
filepath = report_generation_service.generate_technique_detail_report(
db, str(technique_id), output_format=format,
)
# Return FileResponse(
return FileResponse(
_assert_safe_report_path(filepath),
# Keyword argument: media_type
media_type=_MEDIA_TYPES[format],
# Keyword argument: filename
filename=f"technique_{technique_id}.{format}",
)