diff --git a/backend/app/routers/professional_reports.py b/backend/app/routers/professional_reports.py index b91ef24..c15e6d1 100644 --- a/backend/app/routers/professional_reports.py +++ b/backend/app/routers/professional_reports.py @@ -2,9 +2,10 @@ # Import UUID from uuid from uuid import UUID +from pathlib import Path -# Import APIRouter, Depends, Query, Request from fastapi -from fastapi import APIRouter, Depends, Query, Request +# 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 @@ -21,12 +22,24 @@ 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"]) @@ -65,7 +78,7 @@ def generate_purple_report( ) # Return FileResponse( return FileResponse( - filepath, + _assert_safe_report_path(filepath), # Keyword argument: media_type media_type=_MEDIA_TYPES[format], # Keyword argument: filename @@ -95,7 +108,7 @@ def generate_coverage_report( ) # Return FileResponse( return FileResponse( - filepath, + _assert_safe_report_path(filepath), # Keyword argument: media_type media_type=_MEDIA_TYPES[format], # Keyword argument: filename @@ -125,7 +138,7 @@ def generate_executive_report( ) # Return FileResponse( return FileResponse( - filepath, + _assert_safe_report_path(filepath), # Keyword argument: media_type media_type=_MEDIA_TYPES[format], # Keyword argument: filename @@ -155,7 +168,7 @@ def generate_quarterly_report( ) # Return FileResponse( return FileResponse( - filepath, + _assert_safe_report_path(filepath), # Keyword argument: media_type media_type=_MEDIA_TYPES[format], # Keyword argument: filename @@ -187,7 +200,7 @@ def generate_technique_report( ) # Return FileResponse( return FileResponse( - filepath, + _assert_safe_report_path(filepath), # Keyword argument: media_type media_type=_MEDIA_TYPES[format], # Keyword argument: filename diff --git a/backend/app/routers/sso.py b/backend/app/routers/sso.py index 4f72a81..db0ca91 100644 --- a/backend/app/routers/sso.py +++ b/backend/app/routers/sso.py @@ -1,6 +1,7 @@ """Phase 14: SSO / SAML 2.0 router.""" import os +from urllib.parse import urlparse from fastapi import APIRouter, Depends, HTTPException, Request, Response from fastapi.responses import RedirectResponse @@ -74,7 +75,10 @@ def sso_login(request: Request, db: Session = Depends(get_db)): result = svc.initiate_login(db, request_data) except RuntimeError as exc: raise HTTPException(status_code=503, detail=str(exc)) - return RedirectResponse(url=result["redirect_url"]) + redirect_url = result["redirect_url"] + if urlparse(redirect_url).scheme not in ("http", "https"): + raise HTTPException(status_code=400, detail="Invalid IdP redirect URL") + return RedirectResponse(url=redirect_url) @router.post("/callback") diff --git a/backend/app/services/caldera_import_service.py b/backend/app/services/caldera_import_service.py index a7eb838..30b4057 100644 --- a/backend/app/services/caldera_import_service.py +++ b/backend/app/services/caldera_import_service.py @@ -102,9 +102,14 @@ def _download_zip(url: str = CALDERA_ZIP_URL) -> bytes: # Define function _extract_zip def _extract_zip(zip_bytes: bytes, dest: str) -> Path: """Extract *zip_bytes* into *dest* and return abilities dir.""" - # Open context manager + dest_path = Path(dest).resolve() with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf: - # Call zf.extractall() + for member in zf.infolist(): + target = (dest_path / member.filename).resolve() + if not target.is_relative_to(dest_path): + raise ValueError( + f"Zip Slip detected — '{member.filename}' resolves outside target directory" + ) zf.extractall(dest) # Assign abilities_dir = Path(dest) / _ZIP_ROOT_PREFIX / "data" / "abilities" abilities_dir = Path(dest) / _ZIP_ROOT_PREFIX / "data" / "abilities" diff --git a/backend/app/services/lolbas_import_service.py b/backend/app/services/lolbas_import_service.py index 213a1c0..ce1be3b 100644 --- a/backend/app/services/lolbas_import_service.py +++ b/backend/app/services/lolbas_import_service.py @@ -151,11 +151,15 @@ def _download_zip(url: str) -> bytes: # Define function _extract_zip def _extract_zip(zip_bytes: bytes, dest: str) -> Path: """Extract *zip_bytes* into *dest* and return the root directory.""" - # Open context manager + dest_path = Path(dest).resolve() with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf: - # Call zf.extractall() + for member in zf.infolist(): + target = (dest_path / member.filename).resolve() + if not target.is_relative_to(dest_path): + raise ValueError( + f"Zip Slip detected — '{member.filename}' resolves outside target directory" + ) zf.extractall(dest) - # Return Path(dest) return Path(dest) diff --git a/backend/app/services/threat_actor_import_service.py b/backend/app/services/threat_actor_import_service.py index 673ad47..15cabaa 100644 --- a/backend/app/services/threat_actor_import_service.py +++ b/backend/app/services/threat_actor_import_service.py @@ -112,9 +112,14 @@ def _download_zip(url: str = MITRE_CTI_ZIP_URL) -> bytes: # Define function _extract_zip_and_load_bundle def _extract_zip_and_load_bundle(zip_bytes: bytes, dest: str) -> dict: """Extract ZIP and load the enterprise-attack STIX bundle.""" - # Open context manager + dest_path = Path(dest).resolve() with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf: - # Call zf.extractall() + for member in zf.infolist(): + target = (dest_path / member.filename).resolve() + if not target.is_relative_to(dest_path): + raise ValueError( + f"Zip Slip detected — '{member.filename}' resolves outside target directory" + ) zf.extractall(dest) # Assign bundle_path = ( diff --git a/frontend/src/pages/CompliancePage.tsx b/frontend/src/pages/CompliancePage.tsx index 29ced56..e875d81 100644 --- a/frontend/src/pages/CompliancePage.tsx +++ b/frontend/src/pages/CompliancePage.tsx @@ -102,9 +102,13 @@ export default function CompliancePage() { const json = JSON.stringify(frameworkStatus, null, 2); const blob = new Blob([json], { type: "application/json" }); const url = URL.createObjectURL(blob); + const safeName = frameworkStatus.framework.name + .replace(/[^a-zA-Z0-9\s_-]/g, "") + .replace(/\s+/g, "_") + .substring(0, 64); const a = document.createElement("a"); a.href = url; - a.download = `compliance_${frameworkStatus.framework.name.replace(/\s+/g, "_")}.json`; + a.download = `compliance_${safeName}.json`; document.body.appendChild(a); a.click(); document.body.removeChild(a); diff --git a/frontend/src/pages/ImportRTPage.tsx b/frontend/src/pages/ImportRTPage.tsx index 6dd1b5c..eb10fbc 100644 --- a/frontend/src/pages/ImportRTPage.tsx +++ b/frontend/src/pages/ImportRTPage.tsx @@ -373,8 +373,12 @@ export default function ImportRTPage() {
{t.evidence.slice(0, 3).map((ev: RTEvidenceEntry, ei: number) => { - const ext = ev.filename.split(".").pop()?.toLowerCase() ?? "png"; - const mime = ext === "jpg" || ext === "jpeg" ? "image/jpeg" : `image/${ext}`; + const SAFE_MIMES: Record = { + png: "image/png", jpg: "image/jpeg", jpeg: "image/jpeg", + gif: "image/gif", webp: "image/webp", + }; + const ext = ev.filename.split(".").pop()?.toLowerCase() ?? ""; + const mime = SAFE_MIMES[ext] ?? "image/png"; return (
{ref.url ? ( bool: diff --git a/scripts/verify_gaps.py b/scripts/verify_gaps.py index fe85ca8..8f119e2 100644 --- a/scripts/verify_gaps.py +++ b/scripts/verify_gaps.py @@ -4,9 +4,11 @@ Verify Phase 13 gap fixes: 2. evaluate_all_rules creates in-app notifications for admins 3. webhook dispatch_webhook_targeted exists and is callable """ +import os import requests, sys -BASE = "http://localhost:8000/api/v1" +BASE = os.environ.get("AEGIS_BASE_URL", "http://localhost:8000/api/v1") +ADMIN_PASSWORD = os.environ.get("AEGIS_ADMIN_PASSWORD", "admin123") PASS = "\033[92m✓\033[0m" FAIL = "\033[91m✗\033[0m" passed = 0 @@ -51,7 +53,7 @@ def main(): # ── Gap 2: in-app notifications dispatched ──────────────────────────────── print("── Gap 2: In-app notifications on alert fire ──") tok = requests.post(f"{BASE}/auth/login", - data={"username": "administrator", "password": "admin123"}) + data={"username": "administrator", "password": ADMIN_PASSWORD}) if tok.status_code != 200: print(f" Login failed: {tok.text}"); sys.exit(1) h = {"Authorization": f"Bearer {tok.json().get('access_token')}"}