feat(classification): org data classification scheme, tactic-based initial default, operator edit rights

This commit is contained in:
kitos
2026-07-07 11:16:35 +02:00
parent 949e5d42c4
commit a66fefa30b
12 changed files with 205 additions and 29 deletions
+11 -8
View File
@@ -91,13 +91,16 @@ class AttackSuccessResult(str, enum.Enum):
# Define class DataClassification
class DataClassification(str, enum.Enum):
"""Data sensitivity classification levels for compliance and retention policies."""
"""Data sensitivity classification levels for compliance and retention policies.
# Assign public = "public"
public = "public"
# Assign internal = "internal"
internal = "internal"
# Assign sensitive = "sensitive"
sensitive = "sensitive"
# Assign restricted = "restricted"
Matches the organization's data protection scheme:
- public_release: approved for external/public release
- general_use: general internal information, no special restriction
- confidential: internal use only, or shared with trusted people
- restricted: trusted people, customer content, and/or PII
"""
public_release = "public_release"
general_use = "general_use"
confidential = "confidential"
restricted = "restricted"
+1 -1
View File
@@ -91,7 +91,7 @@ class Campaign(Base):
# Assign created_at = Column(DateTime(timezone=True), server_default=func.now())
created_at = Column(DateTime(timezone=True), server_default=func.now())
# Assign data_classification = Column(String(20), nullable=False, server_default="internal")
data_classification = Column(String(20), nullable=False, server_default="internal")
data_classification = Column(String(20), nullable=False, server_default="confidential")
# Recurring scheduling fields
is_recurring = Column(Boolean, default=False)
+1 -1
View File
@@ -51,7 +51,7 @@ class Evidence(Base):
# Assign notes = Column(Text, nullable=True)
notes = Column(Text, nullable=True)
# Assign data_classification = Column(String(20), nullable=False, server_default="internal")
data_classification = Column(String(20), nullable=False, server_default="internal")
data_classification = Column(String(20), nullable=False, server_default="confidential")
# Relationships
test = relationship("Test", back_populates="evidences")
+1 -1
View File
@@ -120,7 +120,7 @@ class Test(Base):
# Assign retest_count = Column(Integer, default=0)
retest_count = Column(Integer, default=0)
# Assign data_classification = Column(String(20), nullable=False, server_default="internal")
data_classification = Column(String(20), nullable=False, server_default="internal")
data_classification = Column(String(20), nullable=False, server_default="confidential")
# ── Relationships ───────────────────────────────────────────────
technique = relationship("Technique", back_populates="tests")
+11 -4
View File
@@ -41,8 +41,8 @@ from sqlalchemy.orm import Session
# Import get_db from app.database
from app.database import get_db
# Import get_current_user, require_any_role, require_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role, require_role
# Import get_current_user, require_any_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role
# Import UnitOfWork from app.domain.unit_of_work
from app.domain.unit_of_work import UnitOfWork
@@ -86,6 +86,7 @@ from app.services.webhook_service import dispatch_webhook
from app.services.test_crud_service import (
create_test as crud_create_test,
)
from app.services.test_crud_service import determine_initial_classification
# Import from app.services.test_crud_service
from app.services.test_crud_service import (
@@ -553,9 +554,14 @@ def update_test_classification(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_role("admin")),
current_user: User = Depends(require_any_role(
"admin", "manager", "red_tech", "red_lead", "blue_tech", "blue_lead",
)),
) -> TestOut:
"""Update the data classification label for a test (admin only).
"""Update the data classification label for a test.
The initial classification is a best-effort default based on the
technique's tactic — any test participant or admin can correct it.
Args:
test_id (uuid.UUID): Primary key of the test to classify.
@@ -1787,6 +1793,7 @@ def import_rt(
# validation_status pending so Blue Lead must confirm
detection_result=detection_result,
blue_validation_status=None,
data_classification=determine_initial_classification(technique),
# Timing
execution_date=exec_date_str,
created_at=datetime.utcnow(),
+3 -3
View File
@@ -37,7 +37,7 @@ class TestCreate(BaseModel):
class TestClassificationUpdate(BaseModel):
"""Admin-only payload for changing data classification."""
"""Payload for changing a test's data classification (any test participant or admin)."""
# data_classification: DataClassification
data_classification: DataClassification
@@ -299,8 +299,8 @@ class TestOut(BaseModel):
retest_of: uuid.UUID | None = None
# Assign retest_count = 0
retest_count: int = 0
# Assign data_classification = "internal"
data_classification: str = "internal"
# Assign data_classification = "confidential"
data_classification: str = "confidential"
# Technique info (populated when joined)
technique_mitre_id: str | None = None
+2
View File
@@ -54,6 +54,7 @@ from app.models.test_template import TestTemplate
# Import User from app.models.user
from app.models.user import User
from app.services.test_crud_service import determine_initial_classification
# Assign logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
@@ -348,6 +349,7 @@ def _seed_tests(db: Session, users: list[User], techniques: list[Technique], cou
state=state,
# Keyword argument: created_at
created_at=datetime.utcnow() - timedelta(days=random.randint(0, 90)),
data_classification=determine_initial_classification(technique),
)
# Populate team fields based on state
@@ -42,6 +42,7 @@ from app.models.test import Test
from app.models.user import User
from app.services.audit_service import log_action
from app.services.status_service import recalculate_technique_status
from app.services.test_crud_service import determine_initial_classification
logger = logging.getLogger(__name__)
@@ -631,6 +632,7 @@ def import_evaluation_round(
red_validated_by=current_user.id,
red_validated_at=datetime.utcnow(),
detection_result=detection_result,
data_classification=determine_initial_classification(technique),
blue_validation_status=None,
execution_date=datetime.utcnow(),
created_at=datetime.utcnow(),
+12
View File
@@ -79,6 +79,7 @@ JIRA_FIELD_ATTACK_SUCCESS = "customfield_11815"
JIRA_FIELD_ATTACK_DETECTED = "customfield_11809"
JIRA_FIELD_TIME_TO_DETECT = "customfield_11810"
JIRA_FIELD_TIME_TO_CONTAIN = "customfield_11811"
JIRA_FIELD_DATA_SENSITIVITY = "customfield_11233"
_ATTACK_SUCCESS_TO_JIRA = {
"successful": "Yes",
@@ -92,6 +93,13 @@ _DETECTION_TO_JIRA = {
"partially_detected": "Partial",
}
_DATA_CLASSIFICATION_TO_JIRA = {
"public_release": "Public Release",
"general_use": "General Use",
"confidential": "Confidential",
"restricted": "Restricted",
}
# Assign logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
@@ -667,6 +675,10 @@ def auto_create_test_issue(
JIRA_FIELD_SEVERITY: severity,
}
data_sensitivity = _DATA_CLASSIFICATION_TO_JIRA.get(_enum_value(test.data_classification))
if data_sensitivity:
fields[JIRA_FIELD_DATA_SENSITIVITY] = data_sensitivity
# Inherit campaign start date if available, otherwise use today
from datetime import date as _date
effective_start = campaign_start_date or _date.today()
+26 -1
View File
@@ -18,7 +18,7 @@ from app.domain.errors import (
EntityNotFoundError,
PermissionViolation,
)
from app.models.enums import TestState
from app.models.enums import DataClassification, TestState
from app.models.technique import Technique
from app.models.test import Test
from app.models.test_template import TestTemplate
@@ -26,6 +26,27 @@ from app.models.campaign import CampaignTest
from app.models.audit import AuditLog
from app.utils import escape_like
# Tactics whose test data is more likely to touch customer content, PII, or
# high-impact operational detail — these get an initial classification of
# 'restricted' rather than the 'confidential' baseline. This is a starting
# point only; the assigned operator or lead can always correct it.
_RESTRICTED_TACTICS = frozenset({
"exfiltration", "collection", "credential-access", "impact",
})
def determine_initial_classification(technique: Technique | None) -> str:
"""Best-effort initial data classification for a new test.
Every security test reveals internal attack/defense posture, so the
baseline is 'confidential' (internal use only) — never public or
general use by default. Escalated to 'restricted' when the technique's
tactic implies customer data, PII, or destructive impact.
"""
if technique and technique.tactic and technique.tactic.lower() in _RESTRICTED_TACTICS:
return DataClassification.restricted.value
return DataClassification.confidential.value
# Define function list_tests
def list_tests(
@@ -121,6 +142,8 @@ def create_test(
# Raise EntityNotFoundError
raise EntityNotFoundError("Technique", str(technique_id))
data_classification = fields.pop("data_classification", None) or determine_initial_classification(technique)
# Assign test = Test(
test = Test(
# Keyword argument: technique_id
@@ -130,6 +153,7 @@ def create_test(
# Keyword argument: state
state=TestState.draft,
created_at=datetime.utcnow(), # explicit — DB column has no server default
data_classification=data_classification,
**fields,
)
# Stage new record(s) for database insertion
@@ -223,6 +247,7 @@ def create_test_from_template(
# Keyword argument: state
state=TestState.draft,
created_at=datetime.utcnow(), # explicit — DB column has no server default
data_classification=determine_initial_classification(technique),
)
# Stage new record(s) for database insertion
db.add(test)