anti/DECNET
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ttp): rich ThreatActor STIX extensions via CustomExtension + CustomObject

Browse Source 
- stix_custom.py: DecnetActorFingerprintExt (@CustomExtension) wrapping
  network_behavior (os_guess/hop_distance/tcp_fingerprint/timing_stats/
  phase_sequence/behavior_class/beacon fields/tool_guesses) and
  protocol_fingerprints (ja3_hashes/hassh_hashes/kex_order_raw/
  ssh_client_banners/tls_cert_sha256/payload_simhashes/c2_endpoints).
  XDecnetBehaveProfile (@CustomObject x-decnet-behave-profile) carrying
  full BEHAVE-SHELL observation envelopes + kd_digraph_simhash.
  FINGERPRINT_EXT_DEF singleton extension-definition SDO.
- Drop legacy flat x_decnet_ja3_hashes / x_decnet_hassh_hashes /
  x_decnet_c2_endpoints (pre-v1, no consumers).
- stix_export: _threat_actor() wired to behavior + observations;
  build_attacker_bundle/build_fleet_bundle grow observations parameter.
- Repo: list_observations_by_attacker + get_all_observations_for_export
  abstract + sqlmodel impl; all four export endpoints extended.
- 18 new tests; inter-DECNET round-trip (stix2.parse → typed objects)
  is the primary fidelity assertion.
This commit is contained in:
anti
2026-05-09 08:52:19 -04:00
parent 1200ac9132
commit 97c99a4e03
16 changed files with 745 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
decnet/ttp/misp_export.py
View File

@@ -47,6 +47,7 @@ def build_attacker_misp_event(
artifacts: list[dict[str, Any]],
smtp_targets: list[dict[str, Any]],
commands: list[str] | None = None,
observations: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Return a MISP event dict for *attacker*.
@@ -63,6 +64,7 @@ def build_attacker_misp_event(
artifacts=artifacts,
smtp_targets=smtp_targets,
commands=commands,
observations=observations,
)
return _parse_bundle(bundle)
@@ -70,6 +72,7 @@ def build_attacker_misp_event(
def build_fleet_misp_collection(
rows: list[dict[str, Any]],
ttp_by_attacker: dict[str, list[dict[str, Any]]],
observations_by_attacker: dict[str, list[dict[str, Any]]] | None = None,
) -> dict[str, Any]:
"""Return a MISP collection dict with one event per attacker in *rows*.
@@ -80,6 +83,7 @@ def build_fleet_misp_collection(
attacker always has at least an IP) are silently omitted.
"""
events: list[dict[str, Any]] = []
obs_map = observations_by_attacker or {}
for row in rows:
raw_cmds = row.get("commands") or []
if isinstance(raw_cmds, str):
@@ -102,6 +106,7 @@ def build_fleet_misp_collection(
artifacts=[],
smtp_targets=[],
commands=cmds,
observations=obs_map.get(row["uuid"]),
)
event = _parse_bundle(bundle)
if event:

100
decnet/ttp/stix_custom.py Normal file
View File

@@ -0,0 +1,100 @@
"""DECNET-defined STIX 2.1 custom extension and object types.
Import this module before parsing any DECNET-produced bundle so the types are
registered with the stix2 library and ``stix2.parse(bundle, allow_custom=True)``
rebuilds them as typed objects rather than opaque dicts.
Classes
-------
DecnetActorFingerprintExt
``@CustomExtension`` on ThreatActor — carries ``network_behavior``
(TCP/TLS/SSH sniffer rollup) and ``protocol_fingerprints`` (hashes +
raw orderings).
XDecnetBehaveProfile
``@CustomObject`` — autonomous STIX SDO carrying the BEHAVE-SHELL
observation stream for one attacker. Referenced from ThreatActor via
``x_decnet_behave_profile_ref``.
Constants
---------
ACTOR_FINGERPRINT_EXT_ID : str
Fixed ``extension-definition--`` ID for ``DecnetActorFingerprintExt``.
FINGERPRINT_EXT_DEF : stix2.ExtensionDefinition
Singleton ``extension-definition`` SDO — add to every bundle that uses
the fingerprint extension.
"""
from __future__ import annotations
import uuid as _uuid
import stix2
from stix2 import CustomExtension, CustomObject, ExtensionDefinition
from stix2 import properties as _P
_NS = _uuid.UUID("b5d2c3a1-8f4e-4d1b-9a6c-0e7f5b3d2c1a")
# Stable ID for the actor fingerprint extension-definition SDO.
ACTOR_FINGERPRINT_EXT_ID: str = (
f"extension-definition--{_uuid.uuid5(_NS, 'decnet-actor-fingerprint-v1')}"
)
_DECNET_ORG_ID = f"identity--{_uuid.uuid5(_NS, 'decnet-honeypot')}"
@CustomExtension(
ACTOR_FINGERPRINT_EXT_ID,
[
("extension_type", _P.StringProperty(required=True)),
("network_behavior", _P.DictionaryProperty()),
("protocol_fingerprints", _P.DictionaryProperty()),
],
)
class DecnetActorFingerprintExt:
"""Property extension on ThreatActor.
``network_behavior`` keys: os_guess, hop_distance, tcp_fingerprint,
retransmit_count, timing_stats, phase_sequence, behavior_class,
beacon_interval_s, beacon_jitter_pct, tool_guesses.
``protocol_fingerprints`` keys: ja3_hashes, hassh_hashes, kex_order_raw,
ssh_client_banners, tls_cert_sha256, payload_simhashes, c2_endpoints.
"""
@CustomObject(
"x-decnet-behave-profile",
[
("schema_version", _P.IntegerProperty()),
("kd_digraph_simhash", _P.StringProperty()),
("observations", _P.ListProperty(_P.DictionaryProperty())),
],
)
class XDecnetBehaveProfile:
"""BEHAVE-SHELL observation stream for one attacker.
``observations`` is a list of BEHAVE envelope dicts with keys:
primitive, value, confidence, window (start_ts/end_ts), source,
evidence_ref, identity_ref (optional).
``schema_version`` matches ``OBSERVATION_SCHEMA_VERSION`` from
decnet_behave_shell.spec.envelope — bump when the envelope schema changes.
``kd_digraph_simhash`` is the 8-byte digraph SimHash from
AttackerIdentity, hex-encoded. Null when identity has not been clustered.
"""
# Singleton extension-definition SDO.
FINGERPRINT_EXT_DEF: stix2.ExtensionDefinition = ExtensionDefinition(
id=ACTOR_FINGERPRINT_EXT_ID,
name="DECNET Actor Fingerprint",
description=(
"Extends ThreatActor with DECNET-observed network behavior "
"(TCP/TLS/SSH stack-level fingerprints, IAT timing, phase sequence) "
"and BEHAVE-SHELL keystroke-dynamics observation primitives."
),
schema="https://decnet.dev/schemas/actor-fingerprint/v1",
version="1.0.0",
extension_types=["property-extension"],
created_by_ref=_DECNET_ORG_ID,
)

117
decnet/ttp/stix_export.py
View File

@@ -23,6 +23,7 @@ public ATT&CK bundle by any consumer that already has it.
"""
from __future__ import annotations
import base64
import json
import uuid as _uuid
from datetime import datetime, timezone
@@ -31,6 +32,12 @@ from typing import Any
import stix2
from decnet.ttp import attack_stix
from decnet.ttp.stix_custom import (
ACTOR_FINGERPRINT_EXT_ID,
FINGERPRINT_EXT_DEF,
DecnetActorFingerprintExt,
XDecnetBehaveProfile,
)
# Deterministic DECNET org identity ID — stable across all bundles this
# instance produces. Consumers can correlate across exports.
@@ -57,15 +64,32 @@ def _decnet_org() -> stix2.Identity:
)
def _parse_json_field(v: Any) -> Any:
"""JSON-decode strings; return non-strings unchanged."""
if isinstance(v, str):
try:
return json.loads(v)
except Exception:
return v
return v
def _threat_actor(
attacker: dict[str, Any],
identity: dict[str, Any] | None,
created_by: str,
) -> stix2.ThreatActor:
behavior: dict[str, Any] | None = None,
observations: list[dict[str, Any]] | None = None,
) -> tuple[stix2.ThreatActor, "XDecnetBehaveProfile | None"]:
"""Build a ThreatActor SDO plus an optional XDecnetBehaveProfile SDO.
Returns ``(threat_actor, behave_profile_or_None)``.
"""
if identity:
name = f"DECNET-identity-{identity['uuid'][:8]}"
else:
name = f"DECNET-attacker-{attacker['uuid'][:8]}"
kwargs: dict[str, Any] = dict(
id=f"threat-actor--{_uuid.uuid5(_NS, attacker['uuid'])}",
name=name,
@@ -73,20 +97,84 @@ def _threat_actor(
created_by_ref=created_by,
allow_custom=True,
)
# Tier 1 — stable scalars
if attacker.get("country_code"):
kwargs["x_decnet_country_code"] = attacker["country_code"]
if attacker.get("asn"):
kwargs["x_decnet_asn"] = attacker["asn"]
if attacker.get("as_name"):
kwargs["x_decnet_as_name"] = attacker["as_name"]
# Tier 2 — DecnetActorFingerprintExt (network_behavior + protocol_fingerprints)
network_behavior: dict[str, Any] = {}
protocol_fingerprints: dict[str, Any] = {}
if behavior:
for key in ("os_guess", "hop_distance", "retransmit_count",
"behavior_class", "beacon_interval_s", "beacon_jitter_pct"):
v = behavior.get(key)
if v is not None:
network_behavior[key] = v
for key in ("tcp_fingerprint", "timing_stats", "phase_sequence", "tool_guesses"):
v = _parse_json_field(behavior.get(key))
if v:
network_behavior[key] = v
for key in ("kex_order_raw", "ssh_client_banners"):
v = _parse_json_field(behavior.get(key))
if v:
protocol_fingerprints[key] = v
if identity:
if identity.get("ja3_hashes"):
kwargs["x_decnet_ja3_hashes"] = identity["ja3_hashes"]
if identity.get("hassh_hashes"):
kwargs["x_decnet_hassh_hashes"] = identity["hassh_hashes"]
for key in ("ja3_hashes", "hassh_hashes", "tls_cert_sha256", "payload_simhashes"):
v = _parse_json_field(identity.get(key))
if v:
protocol_fingerprints[key] = v
if identity.get("c2_endpoints"):
kwargs["x_decnet_c2_endpoints"] = identity["c2_endpoints"]
return stix2.ThreatActor(**kwargs)
protocol_fingerprints["c2_endpoints"] = _parse_json_field(
identity["c2_endpoints"]
)
if network_behavior or protocol_fingerprints:
ext_kwargs: dict[str, Any] = {"extension_type": "property-extension"}
if network_behavior:
ext_kwargs["network_behavior"] = network_behavior
if protocol_fingerprints:
ext_kwargs["protocol_fingerprints"] = protocol_fingerprints
kwargs["extensions"] = {
ACTOR_FINGERPRINT_EXT_ID: DecnetActorFingerprintExt(**ext_kwargs),
}
# Tier 3 — XDecnetBehaveProfile (BEHAVE observations)
behave_profile: XDecnetBehaveProfile | None = None
kd_hash: str | None = None
if identity:
raw_kd = identity.get("kd_digraph_simhash")
if raw_kd is not None:
if isinstance(raw_kd, (bytes, bytearray)):
kd_hash = raw_kd.hex()
elif isinstance(raw_kd, str) and raw_kd:
try:
kd_hash = base64.b64decode(raw_kd).hex()
except Exception:
kd_hash = raw_kd
obs_list = observations or []
if obs_list or kd_hash is not None:
from decnet_behave_shell.spec.envelope import OBSERVATION_SCHEMA_VERSION
profile_id = (
f"x-decnet-behave-profile--{_uuid.uuid5(_NS, attacker['uuid'])}"
)
behave_profile = XDecnetBehaveProfile( # type: ignore[call-arg]
id=profile_id,
created_by_ref=created_by,
schema_version=OBSERVATION_SCHEMA_VERSION,
kd_digraph_simhash=kd_hash,
observations=obs_list,
)
kwargs["x_decnet_behave_profile_ref"] = profile_id
return stix2.ThreatActor(**kwargs), behave_profile
def _attack_pattern_sdo(technique_id: str, created_by: str) -> stix2.AttackPattern | None:
@@ -157,6 +245,7 @@ def build_attacker_bundle(
artifacts: list[dict[str, Any]],
smtp_targets: list[dict[str, Any]],
commands: list[str] | None = None,
observations: list[dict[str, Any]] | None = None,
) -> stix2.Bundle:
"""Assemble a STIX 2.1 Bundle for *attacker*.
@@ -185,9 +274,16 @@ def build_attacker_bundle(
)
objs.append(ip_obs)
# ── Threat actor ─────────────────────────────────────────────────
ta = _threat_actor(attacker, identity, org.id)
# ── Threat actor + BEHAVE profile ────────────────────────────────
ta, behave_profile = _threat_actor(
attacker, identity, org.id,
behavior=behavior,
observations=observations,
)
objs.append(ta)
if behave_profile is not None:
objs.append(behave_profile)
objs.append(FINGERPRINT_EXT_DEF)
# ── ATT&CK — attack-patterns + uses relationships + sightings ───
# Build per-technique once; sightings reference the same AP STIX ID.
@@ -315,6 +411,7 @@ def build_attacker_bundle(
def build_fleet_bundle(
rows: list[dict[str, Any]],
ttp_by_attacker: dict[str, list[dict[str, Any]]],
observations_by_attacker: dict[str, list[dict[str, Any]]] | None = None,
) -> stix2.Bundle:
"""Assemble a STIX 2.1 Bundle covering all attackers in *rows*.
@@ -324,6 +421,7 @@ def build_fleet_bundle(
(too verbose; use the per-attacker endpoint for full fidelity).
"""
objs_by_id: dict[str, Any] = {}
obs_map = observations_by_attacker or {}
for row in rows:
raw_cmds = row.get("commands") or []
@@ -349,6 +447,7 @@ def build_fleet_bundle(
artifacts=[],
smtp_targets=[],
commands=cmds,
observations=obs_map.get(row["uuid"]),
)
for obj in bundle.objects:
objs_by_id[obj.id] = obj