kitos
135 lines
4.8 KiB
Python
135 lines
4.8 KiB
Python
"""Technique query service — framework-agnostic queries for technique details."""
|
|
# Enable future language features for compatibility
|
|
from __future__ import annotations
|
|
|
|
# Import Session, joinedload from sqlalchemy.orm
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
# Import EntityNotFoundError from app.domain.errors
|
|
from app.domain.errors import EntityNotFoundError
|
|
|
|
# Import Technique from app.models.technique
|
|
from app.models.technique import Technique
|
|
from app.models.detection_rule import DetectionRule
|
|
from app.models.intel import IntelItem
|
|
from app.services.d3fend_import_service import get_defenses_for_technique
|
|
|
|
# Severity sort order for detection rules (most critical first)
|
|
_SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "informational": 4, None: 5}
|
|
|
|
|
|
# Define function get_technique_detail
|
|
def get_technique_detail(db: Session, mitre_id: str) -> dict:
|
|
"""Fetch full technique details including tests, detection rules, and D3FEND defenses."""
|
|
technique = (
|
|
db.query(Technique)
|
|
# Chain .options() call
|
|
.options(joinedload(Technique.tests))
|
|
# Chain .filter() call
|
|
.filter(Technique.mitre_id == mitre_id)
|
|
# Chain .first() call
|
|
.first()
|
|
)
|
|
# Check: technique is None
|
|
if technique is None:
|
|
# Raise EntityNotFoundError
|
|
raise EntityNotFoundError("Technique", mitre_id)
|
|
|
|
defenses = get_defenses_for_technique(db, technique.id)
|
|
|
|
# Recent intel items for this technique (newest 20)
|
|
intel_items = (
|
|
db.query(IntelItem)
|
|
.filter(IntelItem.technique_id == technique.id)
|
|
.order_by(IntelItem.detected_at.desc())
|
|
.limit(20)
|
|
.all()
|
|
)
|
|
|
|
detection_rules = (
|
|
db.query(DetectionRule)
|
|
.filter(
|
|
DetectionRule.mitre_technique_id == mitre_id,
|
|
DetectionRule.is_active == True, # noqa: E712
|
|
)
|
|
.all()
|
|
)
|
|
# Sort by severity (critical first), then alphabetically by title
|
|
detection_rules.sort(
|
|
key=lambda r: (_SEVERITY_ORDER.get(r.severity, 5), (r.title or "").lower())
|
|
)
|
|
|
|
return {
|
|
# Literal argument value
|
|
"id": str(technique.id),
|
|
# Literal argument value
|
|
"mitre_id": technique.mitre_id,
|
|
# Literal argument value
|
|
"name": technique.name,
|
|
# Literal argument value
|
|
"description": technique.description,
|
|
# Literal argument value
|
|
"tactic": technique.tactic,
|
|
# Literal argument value
|
|
"platforms": technique.platforms or [],
|
|
# Literal argument value
|
|
"mitre_version": technique.mitre_version,
|
|
# Literal argument value
|
|
"mitre_last_modified": technique.mitre_last_modified,
|
|
# Literal argument value
|
|
"is_subtechnique": technique.is_subtechnique,
|
|
# Literal argument value
|
|
"parent_mitre_id": technique.parent_mitre_id,
|
|
# Literal argument value
|
|
"status_global": technique.status_global.value if technique.status_global else "not_evaluated",
|
|
# Literal argument value
|
|
"review_required": technique.review_required,
|
|
# Literal argument value
|
|
"last_review_date": technique.last_review_date,
|
|
# Literal argument value
|
|
"tests": [
|
|
{
|
|
# Literal argument value
|
|
"id": str(t.id),
|
|
# Literal argument value
|
|
"name": t.name,
|
|
# Literal argument value
|
|
"state": t.state.value if t.state else None,
|
|
# Literal argument value
|
|
"result": t.result.value if t.result else None,
|
|
# Literal argument value
|
|
"platform": t.platform,
|
|
# Literal argument value
|
|
"created_at": t.created_at.isoformat() if t.created_at else None,
|
|
}
|
|
for t in technique.tests
|
|
],
|
|
"detection_rules": [
|
|
{
|
|
"id": str(r.id),
|
|
"title": r.title,
|
|
"description": r.description,
|
|
"source": r.source,
|
|
"source_id": r.source_id,
|
|
"source_url": r.source_url,
|
|
"rule_format": r.rule_format,
|
|
"severity": r.severity,
|
|
"platforms": r.platforms or [],
|
|
"false_positive_rate": r.false_positive_rate,
|
|
}
|
|
for r in detection_rules
|
|
],
|
|
"d3fend_defenses": defenses,
|
|
"intel_items": [
|
|
{
|
|
"id": str(item.id),
|
|
"url": item.url,
|
|
"title": item.title,
|
|
"source": item.source,
|
|
"detected_at": item.detected_at.isoformat() if item.detected_at else None,
|
|
"reviewed": item.reviewed,
|
|
}
|
|
for item in intel_items
|
|
],
|
|
}
|