feat(phase-38): automatic intelligence — OSINT enrichment + stale coverage detection
Tarea 4.1 — OSINT Enrichment:
- Add OsintItem model with source_type, severity, CVSS metadata, review flag
- Add Alembic migration b022 with osint_items table and optimized indexes
- Add osint_enrichment_service with NVD API integration, deduplication, rate limiting
- Add OSINT router: GET /osint/items, /osint/summary, /osint/technique/{id}
- Add POST /osint/items/{id}/review to mark items as reviewed
- Add POST /osint/enrich/{technique_id} for manual single-technique enrichment
- Techniques with new CVEs are automatically flagged review_required=True
- Register weekly enrichment job in APScheduler
- Add NVD_API_KEY config setting for optional increased rate limits
Tarea 4.2 — Stale Coverage Detection:
- Add stale_detection_service that flags techniques with no validated test
in the last N days, or never-validated but with a coverage status
- Configurable threshold via STALE_THRESHOLD_DAYS setting (default 365)
- Register daily stale detection job in APScheduler
- Only flags techniques not already marked review_required
This commit is contained in:
197
backend/app/routers/osint.py
Normal file
197
backend/app/routers/osint.py
Normal file
@@ -0,0 +1,197 @@
|
||||
"""OSINT enrichment endpoints — view, review, and trigger enrichment of
|
||||
OSINT items (CVEs, advisories, etc.) linked to techniques.
|
||||
"""
|
||||
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, Query, HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import get_db
|
||||
from app.dependencies.auth import get_current_user, require_any_role
|
||||
from app.models.osint_item import OsintItem
|
||||
from app.models.technique import Technique
|
||||
from app.models.user import User
|
||||
from app.services.osint_enrichment_service import (
|
||||
enrich_technique_with_cves,
|
||||
get_osint_items_for_technique,
|
||||
mark_osint_reviewed,
|
||||
get_unreviewed_count,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/osint", tags=["osint"])
|
||||
|
||||
|
||||
# ── Schemas ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class OsintItemOut(BaseModel):
|
||||
id: str
|
||||
technique_id: str
|
||||
source_type: str
|
||||
source_url: str
|
||||
title: str
|
||||
description: str | None
|
||||
severity: str | None
|
||||
discovered_at: str | None
|
||||
reviewed: bool
|
||||
metadata_: dict | None = None
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
# ── Endpoints ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@router.get("/items")
|
||||
def list_osint_items(
|
||||
technique_id: UUID | None = Query(None),
|
||||
source_type: str | None = Query(None),
|
||||
reviewed: bool | None = Query(None),
|
||||
offset: int = Query(0, ge=0),
|
||||
limit: int = Query(50, ge=1, le=200),
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
):
|
||||
"""List OSINT items with optional filters."""
|
||||
query = db.query(OsintItem)
|
||||
if technique_id:
|
||||
query = query.filter(OsintItem.technique_id == technique_id)
|
||||
if source_type:
|
||||
query = query.filter(OsintItem.source_type == source_type)
|
||||
if reviewed is not None:
|
||||
query = query.filter(OsintItem.reviewed == reviewed)
|
||||
|
||||
total = query.count()
|
||||
items = (
|
||||
query.order_by(OsintItem.discovered_at.desc())
|
||||
.offset(offset)
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
|
||||
return {
|
||||
"total": total,
|
||||
"items": [
|
||||
{
|
||||
"id": str(item.id),
|
||||
"technique_id": str(item.technique_id),
|
||||
"source_type": item.source_type,
|
||||
"source_url": item.source_url,
|
||||
"title": item.title,
|
||||
"description": item.description,
|
||||
"severity": item.severity,
|
||||
"discovered_at": item.discovered_at.isoformat() if item.discovered_at else None,
|
||||
"reviewed": item.reviewed,
|
||||
"metadata": item.metadata_,
|
||||
}
|
||||
for item in items
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@router.get("/summary")
|
||||
def osint_summary(
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Summary statistics for OSINT items."""
|
||||
from sqlalchemy import func
|
||||
|
||||
total = db.query(func.count(OsintItem.id)).scalar() or 0
|
||||
unreviewed = get_unreviewed_count(db)
|
||||
|
||||
by_severity = dict(
|
||||
db.query(OsintItem.severity, func.count(OsintItem.id))
|
||||
.group_by(OsintItem.severity)
|
||||
.all()
|
||||
)
|
||||
|
||||
by_type = dict(
|
||||
db.query(OsintItem.source_type, func.count(OsintItem.id))
|
||||
.group_by(OsintItem.source_type)
|
||||
.all()
|
||||
)
|
||||
|
||||
techniques_with_items = (
|
||||
db.query(func.count(func.distinct(OsintItem.technique_id))).scalar() or 0
|
||||
)
|
||||
|
||||
return {
|
||||
"total_items": total,
|
||||
"unreviewed": unreviewed,
|
||||
"techniques_with_items": techniques_with_items,
|
||||
"by_severity": by_severity,
|
||||
"by_type": by_type,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/items/{item_id}/review")
|
||||
def review_osint_item(
|
||||
item_id: UUID,
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Mark an OSINT item as reviewed."""
|
||||
item = mark_osint_reviewed(db, str(item_id))
|
||||
if not item:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="OSINT item not found",
|
||||
)
|
||||
return {"id": str(item.id), "reviewed": True}
|
||||
|
||||
|
||||
@router.post("/enrich/{technique_id}")
|
||||
def trigger_technique_enrichment(
|
||||
technique_id: UUID,
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(require_any_role("red_lead", "blue_lead")),
|
||||
):
|
||||
"""Manually trigger OSINT enrichment for a single technique."""
|
||||
technique = db.query(Technique).filter(Technique.id == technique_id).first()
|
||||
if not technique:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Technique not found",
|
||||
)
|
||||
|
||||
count = enrich_technique_with_cves(db, technique)
|
||||
return {
|
||||
"technique_id": str(technique.id),
|
||||
"mitre_id": technique.mitre_id,
|
||||
"new_items": count,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/technique/{technique_id}")
|
||||
def get_technique_osint(
|
||||
technique_id: UUID,
|
||||
source_type: str | None = Query(None),
|
||||
reviewed: bool | None = Query(None),
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Get all OSINT items for a specific technique."""
|
||||
items = get_osint_items_for_technique(
|
||||
db,
|
||||
str(technique_id),
|
||||
source_type=source_type,
|
||||
reviewed=reviewed,
|
||||
)
|
||||
return [
|
||||
{
|
||||
"id": str(item.id),
|
||||
"source_type": item.source_type,
|
||||
"source_url": item.source_url,
|
||||
"title": item.title,
|
||||
"description": item.description,
|
||||
"severity": item.severity,
|
||||
"discovered_at": item.discovered_at.isoformat() if item.discovered_at else None,
|
||||
"reviewed": item.reviewed,
|
||||
"metadata": item.metadata_,
|
||||
}
|
||||
for item in items
|
||||
]
|
||||
Reference in New Issue
Block a user