Add system.roles, segment.exposure, proxy_access format (Phases 1-3)

DavidJBianco · claude · DavidJBianco · commit 0aa35286c47d · 2026-03-24T16:00:38.000-04:00
Model additions:
- System.roles: list field for system functions (forward_proxy,
  web_server, dns_server, mail_server, etc.)
- NetworkSegment.exposure: internal|external|both for client IP mix
- Migrated web_server detection from service-name matching to roles

Web access fix:
- Client IPs now based on segment exposure: internal segments get
  only internal clients, external/both segments get mixed traffic
  with ~60% external IPs for internet-facing servers

Proxy access format (infrastructure only, no traffic generation yet):
- New format definition: W3C Extended Log Format with cache result,
  CONNECT support, full URL, authenticated username
- New emitter: ProxyEmitter (HostMultiplexEmitter, per-proxy routing)
- New eval parser: ProxyAccessParser for W3C Extended format
- Registered in engine, validation, and emitter __init__

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/evidenceforge/evaluation/parsers/proxy.py b/src/evidenceforge/evaluation/parsers/proxy.py
@@ -0,0 +1,88 @@
+"""Parser for HTTP/HTTPS forward proxy access logs (W3C Extended format)."""
+
+import re
+from collections.abc import Iterator
+from datetime import datetime, timezone
+from pathlib import Path
+
+from . import LogParser, ParsedRecord, register_parser
+
+# W3C Extended format:
+# date time c-ip cs-username cs-method cs-uri sc-status sc-bytes cs-bytes time-taken "cs(User-Agent)" cs-host rs(Content-Type) s-cache-result
+_PROXY_PATTERN = re.compile(
+    r'^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+'  # timestamp
+    r'(\S+)\s+'                                       # client_ip
+    r'(\S+)\s+'                                       # username
+    r'(\S+)\s+'                                       # method
+    r'(\S+)\s+'                                       # url
+    r'(\d+)\s+'                                       # status_code
+    r'(\d+)\s+'                                       # sc_bytes
+    r'(\d+)\s+'                                       # cs_bytes
+    r'(\d+)\s+'                                       # time_taken
+    r'"([^"]*)"\s+'                                   # user_agent
+    r'(\S+)\s+'                                       # host
+    r'(\S+)\s+'                                       # content_type
+    r'(\S+)'                                          # cache_result
+)
+
+
+@register_parser
+class ProxyAccessParser(LogParser):
+    format_name = "proxy_access"
+
+    def can_parse(self, path: Path) -> bool:
+        return path.name == "proxy_access.log"
+
+    def parse_file(self, path: Path) -> Iterator[ParsedRecord]:
+        with open(path) as f:
+            for i, line in enumerate(f, 1):
+                line = line.strip()
+                if not line or line.startswith('#'):
+                    continue
+                yield self._parse_line(line, i)
+
+    def _parse_line(self, line: str, line_number: int) -> ParsedRecord:
+        fields: dict = {}
+        errors: list[str] = []
+        timestamp = None
+
+        match = _PROXY_PATTERN.match(line)
+        if not match:
+            errors.append(f"Line does not match proxy access format")
+            return ParsedRecord(
+                source_format=self.format_name,
+                raw=line, fields=fields,
+                timestamp=None, parse_errors=errors,
+                line_number=line_number,
+            )
+
+        ts_str = match.group(1)
+        try:
+            timestamp = datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
+        except ValueError:
+            errors.append(f"Invalid timestamp: {ts_str}")
+
+        fields["timestamp"] = ts_str
+        fields["client_ip"] = match.group(2)
+        username = match.group(3)
+        fields["username"] = username if username != '-' else None
+        fields["method"] = match.group(4)
+        fields["url"] = match.group(5)
+        fields["status_code"] = int(match.group(6))
+        fields["sc_bytes"] = int(match.group(7))
+        fields["cs_bytes"] = int(match.group(8))
+        fields["time_taken"] = int(match.group(9))
+        ua = match.group(10)
+        fields["user_agent"] = ua if ua != '-' else None
+        fields["host"] = match.group(11)
+        ct = match.group(12)
+        fields["content_type"] = ct if ct != '-' else None
+        cr = match.group(13)
+        fields["cache_result"] = cr if cr != '-' else None
+
+        return ParsedRecord(
+            source_format=self.format_name,
+            raw=line, fields=fields,
+            timestamp=timestamp, parse_errors=errors,
+            line_number=line_number,
+        )
diff --git a/src/evidenceforge/formats/definitions/proxy_access.yaml b/src/evidenceforge/formats/definitions/proxy_access.yaml
@@ -0,0 +1,89 @@
+name: proxy_access
+version: "1.0"
+description: "HTTP/HTTPS forward proxy access log (W3C Extended Log Format)"
+category: host
+
+fields:
+  - name: timestamp
+    type: timestamp
+    required: true
+    description: "Request timestamp"
+
+  - name: client_ip
+    type: ip_address
+    required: true
+    description: "Original client IP address (internal user)"
+
+  - name: username
+    type: string
+    required: false
+    description: "Authenticated proxy username"
+
+  - name: method
+    type: string
+    required: true
+    description: "HTTP method (GET, POST, CONNECT, HEAD, PUT, DELETE)"
+
+  - name: url
+    type: string
+    required: true
+    description: "Full destination URL or host:port for CONNECT"
+
+  - name: protocol
+    type: string
+    required: true
+    description: "HTTP protocol version"
+
+  - name: status_code
+    type: integer
+    required: true
+    description: "HTTP response status code"
+    constraints:
+      min_value: 100
+      max_value: 599
+
+  - name: sc_bytes
+    type: integer
+    required: false
+    description: "Server to client bytes"
+
+  - name: cs_bytes
+    type: integer
+    required: false
+    description: "Client to server bytes"
+
+  - name: time_taken
+    type: integer
+    required: false
+    description: "Request duration in milliseconds"
+
+  - name: user_agent
+    type: string
+    required: false
+    description: "Client User-Agent string"
+
+  - name: host
+    type: string
+    required: true
+    description: "Destination hostname"
+
+  - name: content_type
+    type: string
+    required: false
+    description: "Response Content-Type"
+
+  - name: cache_result
+    type: string
+    required: false
+    description: "Cache result (HIT, MISS, NONE, DENIED)"
+
+output:
+  format: text
+  file_extension: ".log"
+  encoding: utf-8
+  header_template: |
+    #Software: EvidenceForge Proxy Log
+    #Version: 1.0
+    #Fields: date time c-ip cs-username cs-method cs-uri sc-status sc-bytes cs-bytes time-taken cs(User-Agent) cs-host rs(Content-Type) s-cache-result
+  template: |
+    {{ timestamp.strftime('%Y-%m-%d %H:%M:%S') }} {{ client_ip }} {{ username if username else '-' }} {{ method }} {{ url }} {{ status_code }} {{ sc_bytes if sc_bytes else 0 }} {{ cs_bytes if cs_bytes else 0 }} {{ time_taken if time_taken else 0 }} "{{ user_agent if user_agent else '-' }}" {{ host }} {{ content_type if content_type else '-' }} {{ cache_result if cache_result else '-' }}
diff --git a/src/evidenceforge/generation/emitters/__init__.py b/src/evidenceforge/generation/emitters/__init__.py
@@ -19,6 +19,7 @@
 from evidenceforge.generation.emitters.ecar import EcarEmitter
 from evidenceforge.generation.emitters.syslog import SyslogEmitter
 from evidenceforge.generation.emitters.bash_history import BashHistoryEmitter
+from evidenceforge.generation.emitters.proxy import ProxyEmitter
 from evidenceforge.generation.emitters.snort import SnortEmitter
 from evidenceforge.generation.emitters.sysmon import SysmonEventEmitter
 from evidenceforge.generation.emitters.web import WebEmitter
@@ -43,6 +44,7 @@
     "EcarEmitter",
     "SyslogEmitter",
     "BashHistoryEmitter",
+    "ProxyEmitter",
     "SnortEmitter",
     "SysmonEventEmitter",
     "WebEmitter",
diff --git a/src/evidenceforge/generation/emitters/proxy.py b/src/evidenceforge/generation/emitters/proxy.py
@@ -0,0 +1,41 @@
+"""HTTP/HTTPS forward proxy access log emitter (W3C Extended format)."""
+
+from typing import Any
+
+from evidenceforge.generation.emitters.host_base import HostMultiplexEmitter
+
+
+class ProxyEmitter(HostMultiplexEmitter):
+    """Emitter for forward proxy access logs (W3C Extended Log Format).
+
+    Per-host FQDN directory routing: each proxy server gets its own access log.
+    """
+
+    _log_filename = "proxy_access.log"
+    _supported_types: set[str] = set()  # raw-only via generate_raw()
+
+    def _dispatch(self, event_data: dict[str, Any]) -> None:
+        """Route proxy access event to per-host file."""
+        rendered = self._render_event(event_data)
+        host_fqdn = event_data.pop('_host_fqdn', '')
+        self.emit_to_host(rendered, host_fqdn)
+
+    def _render_event(self, event_data: dict[str, Any]) -> str:
+        """Render proxy access log entry in W3C Extended format."""
+        context = {
+            'timestamp': event_data.get('timestamp'),
+            'client_ip': event_data.get('client_ip'),
+            'username': event_data.get('username'),
+            'method': event_data.get('method'),
+            'url': event_data.get('url'),
+            'status_code': event_data.get('status_code'),
+            'sc_bytes': event_data.get('sc_bytes'),
+            'cs_bytes': event_data.get('cs_bytes'),
+            'time_taken': event_data.get('time_taken'),
+            'user_agent': event_data.get('user_agent'),
+            'host': event_data.get('host'),
+            'content_type': event_data.get('content_type'),
+            'cache_result': event_data.get('cache_result'),
+        }
+        rendered = self._template.render(**context)
+        return rendered.strip()
diff --git a/src/evidenceforge/generation/engine.py b/src/evidenceforge/generation/engine.py
@@ -33,6 +33,7 @@
     EcarEmitter,
     SyslogEmitter,
     BashHistoryEmitter,
+    ProxyEmitter,
     SnortEmitter,
     SysmonEventEmitter,
     WebEmitter,
@@ -211,6 +212,7 @@ def _initialize(self) -> None:
             'bash_history': BashHistoryEmitter,
             'snort_alert': SnortEmitter,
             'web_access': WebEmitter,
+            'proxy_access': ProxyEmitter,
         }
 
         # Build per-format sensor hostname mapping (expand group names)
@@ -224,7 +226,7 @@ def _initialize(self) -> None:
         _ZEEK_FORMATS = {k for k in emitter_classes if k.startswith("zeek_")}
         # Network sensor formats get per-sensor dirs; host-based formats get per-host FQDN dirs
         _SENSOR_FORMATS = _ZEEK_FORMATS | {'snort_alert'}
-        _HOST_FORMATS = {'windows_event_security', 'windows_event_sysmon', 'ecar', 'syslog', 'bash_history', 'web_access'}
+        _HOST_FORMATS = {'windows_event_security', 'windows_event_sysmon', 'ecar', 'syslog', 'bash_history', 'web_access', 'proxy_access'}
 
         for format_name in sorted(formats_to_generate):
             if format_name not in emitter_classes:
@@ -1279,6 +1281,36 @@ def _execute_typed_event(
 
         return malicious_event
 
+    def _get_system_exposure(self, system) -> str:
+        """Get the network exposure for a system based on its segment.
+
+        Returns 'internal', 'external', or 'both'. Defaults to 'both' if
+        no network config exists (backward compat).
+        """
+        if not self.scenario.environment.network:
+            return 'both'
+        import ipaddress as _ipa
+        sys_ip = _ipa.ip_address(system.ip)
+        for seg in self.scenario.environment.network.segments:
+            net = _ipa.ip_network(seg.cidr, strict=False)
+            if sys_ip in net:
+                return seg.exposure
+        return 'internal'
+
+    @staticmethod
+    def _generate_external_client_ip(rng) -> str:
+        """Generate a random external (non-RFC1918) IP for web server clients."""
+        while True:
+            ip = f'{rng.randint(1, 223)}.{rng.randint(0, 255)}.{rng.randint(0, 255)}.{rng.randint(1, 254)}'
+            first = int(ip.split('.')[0])
+            if first == 10 or first == 127:
+                continue
+            if ip.startswith('172.') and 16 <= int(ip.split('.')[1]) <= 31:
+                continue
+            if ip.startswith('192.168.'):
+                continue
+            return ip
+
     def _make_domain_sid(self, rid: int | None = None) -> str:
         """Generate a SID using the scenario's domain SID prefix."""
         for sid in self.activity_generator.sid_registry.values():
@@ -2554,7 +2586,7 @@ def _generate_system_traffic(self, current_hour: datetime) -> None:
                         },
                     )
 
-        # Web access logs: 10-30 requests per hour on systems with web services
+        # Web access logs: 10-30 requests per hour on systems with web_server role
         if 'web_access' in self.emitters:
             _WEB_PATHS = [
                 ('/', 'GET', 200), ('/index.html', 'GET', 200),
@@ -2572,15 +2604,26 @@ def _generate_system_traffic(self, current_hour: datetime) -> None:
                 'python-requests/2.31.0',
             ]
             for sys_obj in systems:
-                if not any(p in svc.lower() for svc in (sys_obj.services or []) for p in ('http', 'iis', 'nginx', 'apache', 'web')):
+                if 'web_server' not in (sys_obj.roles or []):
                     continue
                 num_reqs = rng.randint(10, 30)
-                other_ips = [s.ip for s in systems if s.ip != sys_obj.ip]
+
+                # Determine client IP pool based on segment exposure
+                internal_ips = [s.ip for s in systems if s.ip != sys_obj.ip]
+                exposure = self._get_system_exposure(sys_obj)
                 for _ in range(num_reqs):
                     offset = rng.randint(0, 3599)
                     ts = current_hour + timedelta(seconds=offset)
                     path, method, status = rng.choice(_WEB_PATHS)
-                    client_ip = rng.choice(other_ips) if other_ips else '10.0.0.1'
+                    if exposure == 'external':
+                        client_ip = self._generate_external_client_ip(rng)
+                    elif exposure == 'both':
+                        if rng.random() < 0.6:
+                            client_ip = self._generate_external_client_ip(rng)
+                        else:
+                            client_ip = rng.choice(internal_ips) if internal_ips else '10.0.0.1'
+                    else:  # internal
+                        client_ip = rng.choice(internal_ips) if internal_ips else '10.0.0.1'
                     self.activity_generator.generate_raw(
                         time=ts, target_format='web_access', system=sys_obj, fields={
                             'timestamp': ts, 'client_ip': client_ip,
diff --git a/src/evidenceforge/models/scenario.py b/src/evidenceforge/models/scenario.py
@@ -118,6 +118,7 @@ class System(BaseModel):
     type: str = Field(..., pattern="^(workstation|server|domain_controller)$")
     assigned_user: str | None = None
     services: list[str] = Field(default_factory=list)
+    roles: list[str] = Field(default_factory=list, description="System roles: forward_proxy, web_server, dns_server, mail_server, etc.")
 
     @field_validator("ip")
     @classmethod
@@ -428,6 +429,7 @@ class NetworkSegment(BaseModel):
     cidr: str
     description: str = ""
     systems: list[str] = Field(default_factory=list)
+    exposure: Literal["internal", "external", "both"] = "internal"
 
     @field_validator("cidr")
     @classmethod
diff --git a/src/evidenceforge/validation/schema.py b/src/evidenceforge/validation/schema.py
@@ -284,7 +284,7 @@ def _validate_event_sequences(self) -> None:
             'zeek_conn', 'zeek_dns', 'zeek_http', 'zeek_ssl', 'zeek_files',
             'zeek_dhcp', 'zeek_ntp', 'zeek_weird', 'zeek_x509',
             'zeek_ocsp', 'zeek_pe', 'zeek_packet_filter', 'zeek_reporter',
-            'ecar', 'syslog', 'bash_history', 'snort_alert', 'web_access',
+            'ecar', 'syslog', 'bash_history', 'snort_alert', 'web_access', 'proxy_access',
         }
 
         for idx, event in enumerate(self.scenario.storyline):
@@ -399,7 +399,7 @@ def _validate_output_formats(self) -> None:
 
         known_output_formats = (
             set(FORMAT_GROUPS.keys())
-            | {"ecar", "syslog", "bash_history", "snort_alert", "web_access"}
+            | {"ecar", "syslog", "bash_history", "snort_alert", "web_access", "proxy_access"}
         )
         _group_members = {}
         for group, members in FORMAT_GROUPS.items():
diff --git a/tests/integration/test_format_definitions.py b/tests/integration/test_format_definitions.py
@@ -401,8 +401,8 @@ def test_load_all_formats(self):
         """Test that all format definitions load successfully."""
         formats = load_all_formats()
 
-        # Phase 1 (2) + Phase 2.2 (5) + Phase 5.3 (1) + Zeek expansion (11) = 19 formats
-        assert len(formats) == 20
+        # Phase 1 (2) + Phase 2.2 (5) + Phase 5.3 (1) + Zeek expansion (11) + proxy_access (1) = 21
+        assert len(formats) == 21
         assert "windows_event_security" in formats
         assert "zeek_conn" in formats
         assert "ecar" in formats
diff --git a/tests/unit/test_zeek_format_accuracy.py b/tests/unit/test_zeek_format_accuracy.py
@@ -560,4 +560,4 @@ def test_all_formats_load(self):
         """All format YAMLs load successfully."""
         from evidenceforge.formats import load_all_formats
         formats = load_all_formats()
-        assert len(formats) == 20
+        assert len(formats) == 21
