Fix Zeek format issues: timestamp precision and network visibility

Addresses two critical issues identified in generated Zeek conn.log output:

1. Timestamp precision: Format timestamps to exactly 6 decimal places
   (microseconds) to match real Zeek logs. Previously, trailing zeros were
   lost during JSON serialization, causing inconsistent precision.

2. Network connection validation: Add comprehensive validation to prevent
   generating connections that network sensors cannot observe:
   - Same source/destination IP (traffic never traverses network)
   - Localhost addresses (127.0.0.0/8)
   - Link-local addresses (169.254.0.0/16)
   - Multicast/reserved addresses (224.0.0.0/4)

Changes:
- src/log_generator/generation/emitters/zeek.py: Format timestamp as
  string with f"{ts.timestamp():.6f}" before JSON serialization
- src/log_generator/generation/activity.py: Add _is_invalid_network_connection()
  helper, validate in generate_connection(), filter destinations in
  execute_baseline_activity(), update DB server IPs to separate subnet
- src/log_generator/generation/engine.py: Add destination validation in
  storyline execution
- tests/unit/test_zeek_format_accuracy.py: Add test_timestamp_precision()
  to prevent regression (issue occurred multiple times)
- TODO.md: Add Network Visibility Architecture to Phase 2.5 (MVP feature,
  full network topology modeling for realistic sensor placement)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: DavidJBianco

TODO.md

@@ -176,7 +176,27 @@
 - [ ] Timezone handling in utilities
 - [ ] Test: Timezone conversions (UTC internal → system timezone output)
 
-### 2.5 Persona-Based Activity Generation
+### 2.5 Network Visibility Architecture
+
+- [ ] Model network topology and sensor placement in scenario schema
+  - [ ] Define network segments (CIDR ranges) in environment
+  - [ ] Specify sensor placement (which segments are monitored)
+  - [ ] Define sensor capabilities (direction: inbound/outbound/bidirectional)
+- [ ] Implement traffic visibility calculation
+  - [ ] Determine if connection would traverse monitored network points
+  - [ ] Validate connections based on network topology
+  - [ ] Skip connections that wouldn't be visible to configured sensors
+- [ ] Update `generation/activity.py` connection logic
+  - [ ] Check if connection would be observable by network sensors
+  - [ ] Consider source/destination network segments
+  - [ ] Apply sensor placement rules
+- [ ] Test: Intra-segment traffic not visible unless sensor on segment
+- [ ] Test: Cross-segment traffic visible if sensor monitors either segment
+- [ ] Test: External traffic visible if sensor monitors perimeter
+
+**Note:** Phase 1 implemented basic IP validation (no localhost, no same src/dst, no link-local/multicast). This phase adds full network topology modeling for realistic sensor placement.
+
+### 2.6 Persona-Based Activity Generation
 
 - [ ] `generation/persona.py` - Persona activity pattern execution
   - [ ] Load persona definitions (from scenario)
@@ -186,7 +206,7 @@
 - [ ] Test: Persona activity patterns match definitions
 - [ ] Test: Temporal distributions look realistic
 
-### 2.6 LLM Integration (Bedrock Client)
+### 2.7 LLM Integration (Bedrock Client)
 
 - [ ] `llm/client.py` - BedrockClient implementation
   - [ ] Chat and complete methods
@@ -200,14 +220,14 @@
 - [ ] Test: Retry logic with mocked failures
 - [ ] Test: Non-retryable errors fail immediately
 
-### 2.7 Medium Dataset Support
+### 2.8 Medium Dataset Support
 
 - [ ] Optimize StateManager for 100K+ events
 - [ ] Memory profiling to ensure <2GB usage
 - [ ] Test: 8-hour, 100-user scenario (target: ~100K events, <10 min generation time)
 - [ ] Test: Memory usage stays under 2GB
 
-### 2.8 Phase 2 Testing & Scenarios
+### 2.9 Phase 2 Testing & Scenarios
 
 - [ ] Integration test: 8-hour scenario with all 5 formats
 - [ ] Create test fixture: `fixtures/scenarios/medium-dataset.yaml` (100 users, 8 hours)

src/log_generator/generation/activity.py

@@ -75,12 +75,57 @@
         '104.26.7.33',     # gitlab.com
     ],
     'connection_db': [
-        '10.0.10.50',      # Internal DB server
-        '192.168.100.25',  # Internal DB server
+        # For internal DB connections, use dedicated DB server IPs in separate subnet
+        # This prevents matching workstation IPs (10.0.10.x) which would create
+        # same source/destination connections that network sensors can't observe
+        '10.0.100.10',     # Internal DB server (separate subnet)
+        '10.0.100.11',     # Internal DB replica
     ],
 }
 
 
+def _is_invalid_network_connection(src_ip: str, dst_ip: str) -> tuple[bool, str]:
+    """Validate that a network connection would be observable by network sensors.
+
+    Network-based data sources like Zeek can only observe traffic that actually
+    traverses the network. This function checks for connections that would never
+    be visible to network sensors.
+
+    Args:
+        src_ip: Source IP address
+        dst_ip: Destination IP address
+
+    Returns:
+        Tuple of (is_invalid, reason). If is_invalid=True, connection should not be generated.
+    """
+    # Check if source and destination are the same
+    if src_ip == dst_ip:
+        return True, f"Source and destination are identical ({src_ip})"
+
+    # Check for localhost addresses (127.0.0.0/8)
+    # Network sensors cannot observe localhost traffic
+    if src_ip.startswith('127.') or dst_ip.startswith('127.'):
+        return True, f"Connection involves localhost address (src={src_ip}, dst={dst_ip})"
+
+    # Check for link-local addresses (169.254.0.0/16)
+    # These are auto-configured and typically not routed
+    if src_ip.startswith('169.254.') or dst_ip.startswith('169.254.'):
+        return True, f"Connection involves link-local address (src={src_ip}, dst={dst_ip})"
+
+    # Check for multicast addresses (224.0.0.0/4)
+    # These require special handling and shouldn't appear in typical conn logs
+    try:
+        src_first_octet = int(src_ip.split('.')[0])
+        dst_first_octet = int(dst_ip.split('.')[0])
+        if src_first_octet >= 224 or dst_first_octet >= 224:
+            return True, f"Connection involves multicast/reserved address (src={src_ip}, dst={dst_ip})"
+    except (ValueError, IndexError):
+        # Invalid IP format - let it pass, will be caught by other validation
+        pass
+
+    return False, ""
+
+
 class ActivityGenerator:
     """Generates specific activity events using StateManager and emitters.
 
@@ -306,6 +351,15 @@ def generate_connection(
         Returns:
             Zeek UID (18-character string)
         """
+        # Validate connection would be observable by network sensors
+        is_invalid, reason = _is_invalid_network_connection(src_ip, dst_ip)
+        if is_invalid:
+            logger.warning(
+                f"Skipping invalid network connection: {src_ip} -> {dst_ip}. "
+                f"Reason: {reason}. Network sensors would not observe this traffic."
+            )
+            return ""  # Return empty UID to indicate skipped connection
+
         src_port = random.randint(49152, 65535)  # Ephemeral port
 
         # Create connection in StateManager
@@ -411,8 +465,21 @@ def execute_baseline_activity(
 
         # Connection activities
         elif activity_type in EXTERNAL_IPS:
-            # Choose random destination IP
-            dst_ip = random.choice(EXTERNAL_IPS[activity_type])
+            # Choose random destination IP (exclude source system's IP)
+            available_destinations = [
+                ip for ip in EXTERNAL_IPS[activity_type]
+                if ip != system.ip
+            ]
+
+            if not available_destinations:
+                # No valid destinations (all IPs match source)
+                logger.debug(
+                    f"Skipping {activity_type} for {system.hostname}: "
+                    f"no valid destination IPs (all match source {system.ip})"
+                )
+                return
+
+            dst_ip = random.choice(available_destinations)
 
             # Set service and port based on activity type
             if activity_type == 'connection_web':

src/log_generator/generation/emitters/zeek.py

@@ -59,12 +59,13 @@ def _render_event(self, event_data: dict[str, Any]) -> str:
             ts = event_data["ts"]
             if isinstance(ts, datetime):
                 # Convert to epoch with microseconds
-                # NOTE: Ensure datetime objects include microseconds for proper precision
-                event_data["ts"] = ts.timestamp()
+                # Format to exactly 6 decimal places to match Zeek format
+                # This prevents loss of trailing zeros during JSON serialization
+                event_data["ts"] = f"{ts.timestamp():.6f}"
             elif isinstance(ts, str):
                 # If string, parse it
                 dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
-                event_data["ts"] = dt.timestamp()
+                event_data["ts"] = f"{dt.timestamp():.6f}"
 
         # Handle dotted field names (id.orig_h, etc.)
         # Zeek template expects 'data' dict for dotted fields

src/log_generator/generation/engine.py

@@ -524,6 +524,14 @@ def _execute_storyline_event(
             dst_port = details.get('dst_port', 443)
             service = details.get('service', 'https')
 
+            # Validate destination is different from source
+            if dst_ip == system.ip:
+                logger.warning(
+                    f"Skipping storyline connection: dst_ip {dst_ip} matches system IP {system.ip}. "
+                    f"Adjusting to external IP."
+                )
+                dst_ip = '198.51.100.10'  # Force to external IP
+
             uid = self.activity_generator.generate_connection(
                 src_ip=system.ip,
                 dst_ip=dst_ip,

tests/unit/test_zeek_format_accuracy.py

@@ -127,3 +127,85 @@ def test_field_types(self):
             assert expected_type in (str, int, float, bool), (
                 f"Field '{field}' has invalid type: {expected_type}"
             )
+
+    def test_timestamp_precision(self):
+        """Verify Zeek timestamps have exactly 6 decimal places (microseconds).
+
+        This test ensures timestamps maintain microsecond precision and don't
+        lose trailing zeros when serialized to JSON. Real Zeek logs always use
+        exactly 6 decimal places for the epoch timestamp.
+        """
+        from datetime import datetime
+        from log_generator.generation.emitters.zeek import ZeekEmitter
+        from log_generator.formats import load_format
+        from pathlib import Path
+        import tempfile
+        import json
+
+        # Create emitter with temporary output file
+        format_def = load_format('zeek_conn')
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
+            output_file = Path(f.name)
+
+        try:
+            emitter = ZeekEmitter(format_def, output_file)
+
+            # Create test event with specific microseconds
+            test_time = datetime(2024, 1, 15, 10, 30, 45, 123456)  # 123456 microseconds
+
+            event_data = {
+                'ts': test_time,
+                'uid': 'C1234567890ABCDE',
+                'id.orig_h': '10.0.10.5',
+                'id.orig_p': 50000,
+                'id.resp_h': '93.184.216.34',
+                'id.resp_p': 443,
+                'proto': 'tcp',
+                'service': 'https',
+                'duration': 1.5,
+                'orig_bytes': 1000,
+                'resp_bytes': 5000,
+                'conn_state': 'SF',
+                'local_orig': True,
+                'local_resp': False,
+                'missed_bytes': 0,
+                'history': 'ShADadfF',
+                'orig_pkts': 10,
+                'orig_ip_bytes': 1400,
+                'resp_pkts': 12,
+                'resp_ip_bytes': 5480,
+                'ip_proto': 6,
+            }
+
+            emitter.emit_event(event_data)
+            emitter.close()
+
+            # Read the generated JSON
+            with open(output_file) as f:
+                line = f.readline()
+                generated = json.loads(line)
+
+            # Verify timestamp has exactly 6 decimal places
+            ts_str = str(generated['ts'])
+
+            # Split at decimal point
+            assert '.' in ts_str, f"Timestamp missing decimal point: {ts_str}"
+            integer_part, decimal_part = ts_str.split('.')
+
+            # Check exactly 6 decimal places
+            assert len(decimal_part) == 6, (
+                f"Timestamp must have exactly 6 decimal places (microseconds), "
+                f"got {len(decimal_part)}: {ts_str}"
+            )
+
+            # Verify it matches expected value
+            expected_ts = test_time.timestamp()
+            actual_ts = float(generated['ts'])
+            assert abs(actual_ts - expected_ts) < 0.000001, (
+                f"Timestamp value mismatch: expected {expected_ts}, got {actual_ts}"
+            )
+
+        finally:
+            # Clean up
+            if output_file.exists():
+                output_file.unlink()