91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
"""Contract tests for the six per-source lifters (E.1.6).
|
|
|
|
Scoped to the contract surface: each lifter is a :class:`TolerantTagger`
|
|
subclass with a non-empty ``HANDLES`` ⊆ :data:`KNOWN_SOURCE_KINDS`,
|
|
unique ``name``, and an empty-list return from ``_tag_impl``. Behavioral
|
|
absence-tolerance assertions from E.2.6 (per-provider null patterns,
|
|
session-without-AttackerBehavior, etc.) are present but xfail-strict
|
|
pending E.3.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from decnet.ttp.base import KNOWN_SOURCE_KINDS, TaggerEvent, TolerantTagger
|
|
from decnet.ttp.impl.behavioral_lifter import BehavioralLifter
|
|
from decnet.ttp.impl.canary_fingerprint_lifter import CanaryFingerprintLifter
|
|
from decnet.ttp.impl.credential_lifter import CredentialLifter
|
|
from decnet.ttp.impl.email_lifter import EmailLifter
|
|
from decnet.ttp.impl.identity_lifter import IdentityLifter
|
|
from decnet.ttp.impl.intel_lifter import IntelLifter
|
|
|
|
ALL_LIFTERS = [
|
|
BehavioralLifter,
|
|
IntelLifter,
|
|
EmailLifter,
|
|
CanaryFingerprintLifter,
|
|
IdentityLifter,
|
|
CredentialLifter,
|
|
]
|
|
|
|
|
|
def _ev(source_kind: str) -> TaggerEvent:
|
|
return TaggerEvent(
|
|
source_kind=source_kind,
|
|
source_id="src1",
|
|
attacker_uuid="att1",
|
|
identity_uuid=None,
|
|
session_id=None,
|
|
decky_id=None,
|
|
payload={},
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("cls", ALL_LIFTERS)
|
|
def test_lifter_subclasses_tolerant_tagger(cls):
|
|
assert issubclass(cls, TolerantTagger)
|
|
|
|
|
|
@pytest.mark.parametrize("cls", ALL_LIFTERS)
|
|
def test_lifter_handles_is_non_empty_frozenset_subset_of_known(cls):
|
|
assert isinstance(cls.HANDLES, frozenset)
|
|
assert cls.HANDLES, f"{cls.__name__}.HANDLES must not be empty"
|
|
assert cls.HANDLES <= KNOWN_SOURCE_KINDS, (
|
|
f"{cls.__name__}.HANDLES contains kinds not in KNOWN_SOURCE_KINDS"
|
|
)
|
|
|
|
|
|
def test_lifter_names_are_unique_and_non_empty():
|
|
names = [cls.name for cls in ALL_LIFTERS]
|
|
assert all(n for n in names), "every lifter needs a non-empty name"
|
|
assert len(set(names)) == len(names), "lifter names must be unique"
|
|
|
|
|
|
@pytest.mark.parametrize("cls", ALL_LIFTERS)
|
|
def test_lifter_tag_returns_empty_list_for_handled_event(cls):
|
|
lifter = cls()
|
|
kind = next(iter(cls.HANDLES))
|
|
out = asyncio.run(lifter.tag(_ev(kind)))
|
|
assert out == []
|
|
|
|
|
|
@pytest.mark.parametrize("cls", ALL_LIFTERS)
|
|
def test_lifter_instantiable(cls):
|
|
# No abstract methods left — concrete subclass must be constructible.
|
|
cls()
|
|
|
|
|
|
# ── E.2.6 deferred absence-tolerance behavior ──────────────────────
|
|
|
|
|
|
@pytest.mark.xfail(strict=True, reason="impl phase E.3 — IntelLifter null patterns")
|
|
def test_e26_intel_lifter_partial_provider_nulls():
|
|
raise AssertionError("not yet implemented")
|
|
|
|
|
|
@pytest.mark.xfail(strict=True, reason="impl phase E.3 — BehavioralLifter empty join")
|
|
def test_e26_behavioral_lifter_no_attacker_behavior_row():
|
|
raise AssertionError("not yet implemented")
|