test(intel): verify OSINT enrichment and stale coverage detection [FASE-4]

This commit is contained in:
2026-05-18 14:50:31 +02:00
parent bdeeed54e1
commit 2ee59d4e18
3 changed files with 218 additions and 6 deletions

View File

@@ -6,18 +6,19 @@ this with a multi-factor, configurable decay model with confidence scores.
"""
import logging
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.config import settings
from app.models.enums import TechniqueStatus, TestState
from app.models.technique import Technique
from app.models.test import Test
logger = logging.getLogger(__name__)
STALE_THRESHOLD_DAYS = getattr(settings, "STALE_THRESHOLD_DAYS", 365)
STALE_THRESHOLD_DAYS = settings.STALE_THRESHOLD_DAYS
def detect_stale_coverage(db: Session) -> int:
@@ -31,15 +32,21 @@ def detect_stale_coverage(db: Session) -> int:
Returns the number of newly-flagged techniques.
"""
cutoff = datetime.utcnow() - timedelta(days=STALE_THRESHOLD_DAYS)
cutoff = datetime.now(timezone.utc) - timedelta(days=STALE_THRESHOLD_DAYS)
last_validated = func.coalesce(
Test.blue_validated_at,
Test.red_validated_at,
Test.created_at,
)
# Subquery: latest validated test date per technique
latest_test = (
db.query(
Test.technique_id,
func.max(Test.created_at).label("last_tested"),
func.max(last_validated).label("last_tested"),
)
.filter(Test.state == "validated")
.filter(Test.state == TestState.validated)
.group_by(Test.technique_id)
.subquery()
)
@@ -55,7 +62,7 @@ def detect_stale_coverage(db: Session) -> int:
)
.filter(
# Only flag techniques that have a real status (not never-evaluated ones)
Technique.status_global != "not_evaluated"
Technique.status_global != TechniqueStatus.not_evaluated
)
.all()
)