Phase 6.3: Complete P2 moderate realism fixes (remaining 5 items)

Completes all 16 P2 items from REALISM_ASSESSMENT-2.md:

- #22: DB connections require scenario-detected servers (no hardcoded
  10.0.100.x fallback)
- #27: SSH source IPs from scenario system IPs (not 10.0.x.x)
- #28: Syslog emitter sorts buffer chronologically before flush
- #31: eCAR/Zeek source port correlation (shared src_port param)
- #24: Storyline events interleaved into baseline hour loop via
  pre-parsed time index + _execute_single_storyline_event()

765 tests passing.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: DavidJBianco

TODO.md

@@ -767,58 +767,40 @@
 
 ### 6.3 P2: Moderate (Polish & Realism)
 
-- [ ] **Add jitter to storyline timestamps** (Assessment #17)
-  - Attack timestamps are exact multiples of 900s/3600s with .000 microseconds
-  - Background events have realistic jitter; the contrast is obvious
-  - Files: `engine.py` (`_execute_storyline_event`)
-- [ ] **Realistic LogonIDs** (Assessment #18)
-  - TargetLogonId values 0x3e7, 0x3e8, 0x3e9... incrementing by 1
-  - Real LSASS generates high-entropy 64-bit values like 0x1A2B3C4D
-  - 0x3e7 is SYSTEM's well-known LogonID but assigned to regular users
-  - Files: `state_manager.py` (logon ID generation)
-- [ ] **Populate real base64 in encoded commands** (Assessment #19)
-  - Literal `<base64_encoded_command>` placeholder never replaced with actual base64
-  - Files: `engine.py` (storyline execution)
-- [ ] **Use realistic public IPs for exfiltration** (Assessment #20)
-  - RFC 5737 documentation IPs (203.0.113.x, 198.51.100.x) are not internet-routable
-  - Files: scenario YAML, `engine.py`
-- [ ] **Add Kerberos auth package + LogonGuids** (Assessment #21)
-  - LogonGuid always null; Kerberos-enabled domains populate this for cross-machine correlation
-  - AuthenticationPackageName never "Kerberos" — should dominate in AD
-  - Files: `activity.py` (`generate_logon`)
-- [ ] **Fix internal DNS IPs and DB ports** (Assessment #22)
-  - db-primary.corp.local resolves to 10.0.100.x but docs say 10.10.100.x; MySQL 3306 instead of SQL 1433
-  - Files: `activity.py` (REVERSE_DNS, EXTERNAL_IPS), `engine.py`
-- [ ] **Fix Zeek `ts` type consistency** (Assessment #23)
-  - String in zeek_conn.json, bare number in zeek_dns.json; should be consistent
-  - Files: `emitters/zeek.py`, `emitters/zeek_dns.py`
-- [ ] **Interleave scenario events chronologically** (Assessment #24)
-  - Attack events appended as separate block at end of ecar.json/zeek_conn.json
-  - Files: `emitters/base.py` or `engine.py` (output ordering)
-- [ ] **Add FQDN to Computer names** (Assessment #25)
-  - Short names like EXEC-WS-04 instead of EXEC-WS-04.corp.meridiancapital.com
-  - Files: `activity.py` (Computer field in Windows events)
-- [ ] **Add RID gaps and computer account SIDs** (Assessment #26)
-  - RIDs 1001-1105 with zero gaps; no computer account RIDs or deleted-object gaps
-  - Files: `engine.py` (SID registry)
-- [ ] **Use SSH source IPs from documented subnets** (Assessment #27, N4)
-  - SSH from 10.0.x.x instead of documented 10.10.x.x topology
-  - Files: `engine.py` (syslog generation), `activity.py`
-- [ ] **Sort syslog chronologically** (Assessment #28)
-  - Within each hour, timestamps jump randomly; real syslog is append-only/ordered
-  - Files: `engine.py` (`_generate_system_traffic` syslog section), `emitters/syslog.py`
-- [ ] **Limit systemd-timesyncd message** (Assessment #29, N5)
-  - "Synchronized for the first time" repeats dozens of times; should appear once per boot
-  - Files: `engine.py` (syslog templates)
-- [ ] **Diversify Zeek history strings** (Assessment #30)
-  - Only ~9 patterns; real Zeek has dozens including retransmission markers (T/t), varied FIN ordering
-  - Files: `activity.py` (TCP_CONN_STATE_DISTRIBUTION)
-- [ ] **Correlate eCAR/Zeek source ports** (Assessment #31)
-  - eCAR FLOW and Zeek conn at same timestamp show different source ports
-  - Files: `activity.py` (`generate_connection`, `_emit_ecar_flow_event`)
-- [ ] **Add SERVFAIL responses** (Assessment #32)
-  - Zero SERVFAIL in 39K DNS records; real environments have ~0.1-0.5%
-  - Files: `activity.py` (`_emit_dns_lookup`)
+- [x] **Add jitter to storyline timestamps** (Assessment #17)
+  - ±30s random jitter + random microseconds; causal ordering enforced
+- [x] **Realistic LogonIDs** (Assessment #18)
+  - Random 32-bit hex values (0x10000-0xFFFFFFFF), well-known values excluded
+- [x] **Populate real base64 in encoded commands** (Assessment #19)
+  - 10 realistic decoded PowerShell commands, proper UTF-16LE base64 encoding
+- [x] **Use realistic public IPs for exfiltration** (Assessment #20)
+  - Real cloud/hosting IPs (DigitalOcean, Linode, Vultr ranges) replace RFC 5737
+- [x] **Add Kerberos auth package + LogonGuids** (Assessment #21)
+  - Type 3: 70% Kerberos/20% NTLM/10% Negotiate; non-null GUID when Kerberos
+- [x] **Fix internal DNS IPs and DB ports** (Assessment #22)
+  - DB connections require scenario-detected servers; no fallback to hardcoded 10.0.100.x
+- [x] **Fix Zeek `ts` type consistency** (Assessment #23)
+  - Both emitters output float (bare number in JSON), not string
+- [x] **Interleave scenario events chronologically** (Assessment #24)
+  - Storyline events injected into baseline hour loop; remaining executed after
+  - Pre-parsed event times by hour; `_execute_single_storyline_event()` for interleaving
+- [x] **Add FQDN to Computer names** (Assessment #25)
+  - Optional `Environment.domain` field; auto-inferred from user emails
+  - All Windows event Computer fields: `hostname.domain`; dynamic NetBIOS domain
+- [x] **Add RID gaps and computer account SIDs** (Assessment #26)
+  - Random 1-5 gaps between user RIDs; well-known RIDs (500-502); computer SIDs
+- [x] **Use SSH source IPs from documented subnets** (Assessment #27)
+  - SSH sources from scenario system IPs, not hardcoded 10.0.x.x
+- [x] **Sort syslog chronologically** (Assessment #28)
+  - SyslogEmitter sorts buffer by timestamp prefix before flush
+- [x] **Limit systemd-timesyncd message** (Assessment #29)
+  - "for the first time" only once per system; subsequent messages varied
+- [x] **Diversify Zeek history strings** (Assessment #30)
+  - 21 TCP + 8 UDP patterns (was 7+4); retransmission markers, varied FIN ordering
+- [x] **Correlate eCAR/Zeek source ports** (Assessment #31)
+  - src_port passed from generate_connection() to _emit_ecar_flow_event()
+- [x] **Add SERVFAIL responses** (Assessment #32)
+  - 0.2% SERVFAIL; multi-answer DNS (2-5 IPs) for external A records
 
 ### 6.4 P3: Minor (Nice-to-Have Improvements)
 

src/evidenceforge/generation/activity.py

@@ -1139,7 +1139,7 @@ def generate_connection(
 
         # Phase 5.2: Emit eCAR FLOW/CONNECT for eCAR-equipped hosts
         # Use src_ip to find the hostname for this connection
-        self._emit_ecar_flow_event(src_ip, dst_ip, dst_port, time, src_ip)
+        self._emit_ecar_flow_event(src_ip, dst_ip, dst_port, time, src_ip, src_port=src_port)
 
         return uid
 
@@ -1686,8 +1686,8 @@ def execute_baseline_activity(
                     service = db['service']
                     dst_port = db['port']
                 else:
-                    service = 'mysql'
-                    dst_port = 3306
+                    # No DB servers detected from scenario; skip DB connection
+                    return
             else:
                 service = None
                 dst_port = 443
@@ -2204,18 +2204,21 @@ def _emit_ecar_registry_event(
     def _emit_ecar_flow_event(
         self, src_ip: str, dst_ip: str, dst_port: int,
         time: datetime, hostname: str, pid: int = -1,
+        src_port: int = 0,
     ) -> None:
         """Emit eCAR FLOW/CONNECT event."""
         if 'ecar' not in self.emitters:
             return
+        if src_port == 0:
+            src_port = _get_rng().randint(49152, 65535)
         self.emitters['ecar'].emit_event({
             'timestamp': time,
             'hostname': hostname,
             'object': 'FLOW',
             'action': 'CONNECT',
             'pid': pid,
             'src_ip': src_ip,
-            'src_port': _get_rng().randint(49152, 65535),
+            'src_port': src_port,
             'dst_ip': dst_ip,
             'dst_port': dst_port,
             'protocol': 'tcp',

src/evidenceforge/generation/emitters/syslog.py

@@ -18,6 +18,12 @@ def emit_event(self, event_data: dict[str, Any]) -> None:
             rendered = self._render_event(event_data)
             self._buffer_event(rendered)
 
+    def flush(self) -> None:
+        """Flush with chronological sorting (syslog is append-only/ordered)."""
+        with self._file_lock:
+            self.buffer.sort()  # ISO timestamp prefix → lexicographic sort works
+            self._flush_unlocked()
+
     def _render_event(self, event_data: dict[str, Any]) -> str:
         """Render syslog event to text format.
 

src/evidenceforge/generation/engine.py

@@ -121,15 +121,21 @@ def generate(self) -> None:
             self._generate_baseline()
             self._report_progress("phase_end", {"phase": "baseline"})
 
-            # Phase 3: Execute storyline events (if present)
+            # Phase 6.3: Execute remaining storyline events not covered by baseline hours
             if self.scenario.storyline:
-                logger.info(f"Executing {len(self.scenario.storyline)} storyline events")
-                self._report_progress("phase_start", {
-                    "phase": "storyline",
-                    "description": f"Executing {len(self.scenario.storyline)} storyline events"
-                })
-                self._execute_storyline()
-                self._report_progress("phase_end", {"phase": "storyline"})
+                remaining = [i for i in range(len(self.scenario.storyline))
+                             if i not in self._storyline_executed]
+                if remaining:
+                    logger.info(f"Executing {len(remaining)} remaining storyline events (outside baseline window)")
+                    self._report_progress("phase_start", {
+                        "phase": "storyline",
+                        "description": f"Executing {len(remaining)} remaining storyline events"
+                    })
+                    for idx in remaining:
+                        self._execute_single_storyline_event(idx)
+                        self._storyline_executed.add(idx)
+                    self._barrier_flush_all_emitters()
+                    self._report_progress("phase_end", {"phase": "storyline"})
         finally:
             # Phase 4: Finalize and close emitters (always, even on error)
             self._report_progress("phase_start", {"phase": "finalize", "description": "Finalizing generation"})
@@ -255,6 +261,20 @@ def _initialize(self) -> None:
         self._system_pids: dict[str, dict[str, int]] = {}  # hostname -> {role: pid}
         self._seed_system_process_trees()
 
+        # Phase 6.3: Pre-parse storyline event times for interleaved generation
+        self._storyline_by_hour: dict[int, list] = {}  # hour_epoch -> list of (time, event_idx)
+        if self.scenario.storyline:
+            for idx, event in enumerate(self.scenario.storyline):
+                event_time = self._parse_storyline_time(event.time)
+                hour_key = int(event_time.replace(minute=0, second=0, microsecond=0).timestamp())
+                self._storyline_by_hour.setdefault(hour_key, []).append((event_time, idx))
+            # Sort each hour's events by time
+            for key in self._storyline_by_hour:
+                self._storyline_by_hour[key].sort()
+            logger.info(f"Pre-parsed {len(self.scenario.storyline)} storyline events across {len(self._storyline_by_hour)} hours")
+
+        self._storyline_executed: set[int] = set()
+
         logger.info("Initialization complete")
 
     def _generate_baseline(self) -> None:
@@ -326,6 +346,13 @@ def _generate_baseline(self) -> None:
             # Phase 5.4: Generate system traffic (DNS, NTP, scheduled tasks)
             self._generate_system_traffic(current_hour)
 
+            # Phase 6.3: Interleave storyline events into this hour
+            hour_key = int(current_hour.timestamp())
+            for event_time, event_idx in self._storyline_by_hour.get(hour_key, []):
+                if event_idx not in self._storyline_executed:
+                    self._execute_single_storyline_event(event_idx)
+                    self._storyline_executed.add(event_idx)
+
             # Phase 5.2: Terminate stale processes
             self._terminate_stale_processes(current_hour)
 
@@ -854,6 +881,39 @@ def _execute_storyline(self) -> None:
             # Barrier flush after each storyline event (ensures event written before proceeding)
             self._barrier_flush_all_emitters()
 
+    def _execute_single_storyline_event(self, event_idx: int) -> None:
+        """Execute a single storyline event by index (used for interleaved generation)."""
+        storyline_event = self.scenario.storyline[event_idx]
+        event_num = event_idx + 1
+
+        # Parse event time with jitter
+        event_time = self._parse_storyline_time(storyline_event.time)
+        jitter_rng = random.Random(hash(f"jitter_{event_num}_{self.scenario.name}"))
+        jitter = timedelta(
+            seconds=jitter_rng.uniform(-30, 30),
+            microseconds=jitter_rng.randint(0, 999999),
+        )
+        event_time = event_time + jitter
+
+        actor = self._find_actor(storyline_event.actor)
+        system = self._find_system(storyline_event.system)
+        if not actor or not system:
+            return
+
+        logger.info(f"Executing interleaved storyline event: {storyline_event.actor} on {storyline_event.system} at {event_time}")
+
+        event_types = self._match_activity_to_events(storyline_event.activity)
+        self.state_manager.set_current_time(event_time)
+
+        for event_type in event_types:
+            malicious_event = self._execute_storyline_event(
+                actor=actor, system=system, time=event_time,
+                event_type=event_type, activity=storyline_event.activity,
+                details=storyline_event.details,
+            )
+            if malicious_event:
+                self.malicious_events.append(malicious_event)
+
     def _parse_storyline_time(self, time_str: str) -> datetime:
         """Parse storyline event time to absolute datetime.
 
@@ -1773,8 +1833,9 @@ def _generate_system_traffic(self, current_hour: datetime) -> None:
                         'message': action,
                     })
                 elif source_roll < 0.80:
-                    # sshd — disconnect/keepalive messages
-                    ip = f'10.0.{rng.randint(0,10)}.{rng.randint(1,254)}'
+                    # sshd — disconnect/keepalive messages (use scenario system IPs)
+                    other_ips = [s.ip for s in self.scenario.environment.systems if s.ip != system.ip]
+                    ip = rng.choice(other_ips) if other_ips else system.ip
                     port = rng.randint(49152, 65535)
                     msgs = [
                         f'Received disconnect from {ip} port {port}:11: disconnected by user',

tests/unit/test_activity.py

@@ -437,16 +437,18 @@ def test_execute_baseline_activity_connection_git(self, activity_gen, test_user,
         assert event_data['id.resp_h'] in EXTERNAL_IPS['connection_git']
 
     def test_execute_baseline_activity_connection_db(self, activity_gen, test_user, test_system, state_manager, mock_emitters):
-        """execute_baseline_activity should handle database connection."""
+        """execute_baseline_activity should handle database connection with detected servers."""
         timestamp = datetime(2024, 1, 15, 10, 0, 0, tzinfo=timezone.utc)
         state_manager.set_current_time(timestamp)
 
+        # Set up scenario-detected DB servers (required for DB connections)
+        activity_gen._db_servers = [{'ip': '10.10.100.20', 'port': 1433, 'service': 'mssql'}]
         activity_gen.execute_baseline_activity(test_user, test_system, timestamp, 'connection_db')
 
         event_data = mock_emitters['zeek_conn'].emit_event.call_args[0][0]
-        assert event_data['service'] == 'mysql'
-        assert event_data['id.resp_p'] == 3306
-        assert event_data['id.resp_h'] in EXTERNAL_IPS['connection_db']
+        assert event_data['service'] == 'mssql'
+        assert event_data['id.resp_p'] == 1433
+        assert event_data['id.resp_h'] == '10.10.100.20'
 
     def test_execute_baseline_activity_connection_excludes_src_ip(self, activity_gen, test_user, state_manager, mock_emitters):
         """execute_baseline_activity should not connect system to itself."""