Solved issue #206

206/app/api/dashboard.py

@@ -0,0 +1,17 @@
+from fastapi import APIRouter, Depends
+from app.core.security import require_admin
+from app.core.security_events import get_events
+
+router = APIRouter()
+
+@router.get("/")
+def dashboard(role=Depends(require_admin)):
+    events = get_events()
+
+    return {
+        "total_events": len(events),
+        "dependency_alerts": [
+            e for e in events if e["type"] == "DEPENDENCY_VULNERABILITY"
+        ],
+        "all_events": events
+    }

206/app/api/dependency_api.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter, Depends, HTTPException
+from app.core.security import require_admin
+from app.core.dependency_scan import run_dependency_scan
+
+router = APIRouter()
+
+@router.get("/dependency-scan")
+def dependency_scan(role=Depends(require_admin)):
+    result = run_dependency_scan()
+
+    if result["status"] == "vulnerable":
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "Vulnerable dependencies detected",
+                "vulnerabilities": result["details"]
+            }
+        )
+
+    return {"status": "clean"}

206/app/api/internal_test.py

@@ -0,0 +1,12 @@
+from fastapi import APIRouter, Depends
+from app.core.security import require_admin
+from app.core.service_test import test_service_encryption
+
+router = APIRouter()
+CERT_PATH = "./certificates/server.crt"
+
+@router.get("/test-communication")
+async def test_internal_service(role_check=Depends(require_admin)):
+    target_url = "https://localhost:8001/internal-data"
+    result = test_service_encryption(target_url, CERT_PATH)
+    return result

206/app/core/dependency_scan.py

@@ -0,0 +1,38 @@
+import subprocess
+from app.core.security_events import log_event
+
+def run_dependency_scan():
+    try:
+        result = subprocess.run(
+            ["pip-audit", "--format", "json"],
+            capture_output=True,
+            text=True
+        )
+
+        if result.returncode != 0:
+            findings = result.stdout.strip()
+
+            log_event(
+                "DEPENDENCY_VULNERABILITY",
+                {
+                    "tool": "pip-audit",
+                    "findings": findings
+                }
+            )
+
+            return {
+                "status": "vulnerable",
+                "details": findings
+            }
+
+        return {
+            "status": "clean",
+            "details": []
+        }
+
+    except Exception as e:
+        log_event(
+            "DEPENDENCY_SCAN_ERROR",
+            {"error": str(e)}
+        )
+        raise

206/app/core/security.py

@@ -0,0 +1,17 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+
+security = HTTPBearer()
+
+def require_admin(
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+):
+    token = credentials.credentials
+
+    if token != "admin":
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Admin access required"
+        )
+
+    return token

206/app/core/security_events.py

@@ -0,0 +1,13 @@
+from datetime import datetime
+
+SECURITY_EVENTS = []
+
+def log_event(event_type: str, details: dict):
+    SECURITY_EVENTS.append({
+        "type": event_type,
+        "details": details,
+        "timestamp": datetime.utcnow().isoformat()
+    })
+
+def get_events():
+    return SECURITY_EVENTS

206/app/core/service_test.py

@@ -0,0 +1,15 @@
+import requests
+from app.core.security_events import log_event
+
+def test_service_encryption(target_url: str, cert_path: str):
+    try:
+        response = requests.get(target_url, verify=cert_path)
+        if response.status_code == 200:
+            log_event("INTERNAL_COMM_SECURE", {"url": target_url, "status": "TLS verified"})
+            return {"status": "secure", "url": target_url}
+        else:
+            log_event("INTERNAL_COMM_ERROR", {"url": target_url, "status": response.status_code})
+            return {"status": "error", "url": target_url, "code": response.status_code}
+    except requests.exceptions.SSLError as e:
+        log_event("INTERNAL_COMM_TLS_ERROR", {"url": target_url, "error": str(e)})
+        return {"status": "tls_error", "url": target_url, "error": str(e)}

206/app/main.py

@@ -0,0 +1,13 @@
+from fastapi import FastAPI
+from app.middleware.ids_middleware import IDSMiddleware
+from app.api.dashboard import router as dashboard_router
+from app.api.dependency_api import router as dependency_router
+from app.api.internal_test import router as internal_test_router
+
+app = FastAPI(title="Security Dashboard")
+
+app.add_middleware(IDSMiddleware)
+
+app.include_router(dashboard_router, prefix="/dashboard")
+app.include_router(dependency_router, prefix="/security")
+app.include_router(internal_test_router, prefix="/security")

206/app/middleware/ids_middleware.py

@@ -0,0 +1,18 @@
+from starlette.middleware.base import BaseHTTPMiddleware
+from fastapi import Request
+from app.core.security_events import log_event
+
+class IDSMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        response = await call_next(request)
+
+        if response.status_code == 403:
+            log_event(
+                "UNAUTHORIZED_ACCESS",
+                {
+                    "path": request.url.path,
+                    "method": request.method
+                }
+            )
+
+        return response