fix(parser): harden auth parsing and adversarial qualification

.github/workflows/ci.yml

@@ -63,7 +63,7 @@ jobs:
           -D LOGLENS_WARNINGS_AS_ERRORS=ON
 
       - name: Build fuzz target
-        run: cmake --build build-fuzz --target fuzz_parser --config RelWithDebInfo
+        run: cmake --build build-fuzz --target fuzz_parser fuzz_pipeline --config RelWithDebInfo
 
       - name: Run bounded parser fuzz smoke
         run: >-
@@ -72,3 +72,14 @@ jobs:
           -max_len=1024
           -timeout=2
           tests/fuzz/corpus/parser
+
+      - name: Generate synthetic sequence seeds
+        run: python3 scripts/generate-adversarial-corpus.py --output build-fuzz/corpus
+
+      - name: Run bounded semantic pipeline fuzz smoke
+        env:
+          ASAN_OPTIONS: detect_leaks=1
+          UBSAN_OPTIONS: halt_on_error=1:print_stacktrace=1
+        run: >-
+          ./build-fuzz/fuzz_pipeline -seed=129130 -runs=2000
+          -max_len=97 -timeout=3 -rss_limit_mb=1024 build-fuzz/corpus/pipeline

CMakeLists.txt

@@ -73,6 +73,11 @@ target_compile_definitions(loglens PRIVATE LOGLENS_VERSION="${LOGLENS_VERSION}")
 
 include(CTest)
 if(BUILD_TESTING)
+    add_executable(test_adversarial_semantics tests/test_adversarial_semantics.cpp)
+    target_link_libraries(test_adversarial_semantics PRIVATE loglens_lib)
+    loglens_enable_warnings(test_adversarial_semantics)
+    add_test(NAME adversarial_semantics COMMAND test_adversarial_semantics)
+
     add_executable(test_parser tests/test_parser.cpp)
     target_link_libraries(test_parser PRIVATE loglens_lib)
     loglens_enable_warnings(test_parser)
@@ -126,6 +131,13 @@ if(BUILD_TESTING)
 
     find_package(Python3 3.9 COMPONENTS Interpreter QUIET)
     if(Python3_Interpreter_FOUND)
+        add_executable(report_adversarial_driver tests/report_adversarial_driver.cpp)
+        target_link_libraries(report_adversarial_driver PRIVATE loglens_lib)
+        loglens_enable_warnings(report_adversarial_driver)
+        add_test(NAME adversarial_reports COMMAND ${Python3_EXECUTABLE}
+            ${CMAKE_CURRENT_SOURCE_DIR}/tests/test_adversarial_reports.py
+            $<TARGET_FILE:report_adversarial_driver>
+            ${CMAKE_CURRENT_BINARY_DIR}/adversarial_report_output)
         add_test(
             NAME episode_candidate_research
             COMMAND
@@ -176,4 +188,11 @@ if(LOGLENS_BUILD_FUZZERS)
     target_compile_options(fuzz_timestamp_day PRIVATE
         -fsanitize=fuzzer,address,undefined -fno-omit-frame-pointer)
     target_link_options(fuzz_timestamp_day PRIVATE -fsanitize=fuzzer,address,undefined)
+
+    add_executable(fuzz_pipeline fuzz/pipeline_fuzz.cpp)
+    target_link_libraries(fuzz_pipeline PRIVATE loglens_fuzz_lib)
+    loglens_enable_warnings(fuzz_pipeline)
+    target_compile_options(fuzz_pipeline PRIVATE
+        -fsanitize=fuzzer,address,undefined -fno-omit-frame-pointer)
+    target_link_options(fuzz_pipeline PRIVATE -fsanitize=fuzzer,address,undefined)
 endif()

fuzz/pipeline_fuzz.cpp

@@ -0,0 +1,9 @@
+#include "../tests/support/qualification_pipeline.hpp"
+
+#include <cstddef>
+#include <cstdint>
+
+extern "C" int LLVMFuzzerTestOneInput(const std::uint8_t* data, std::size_t size) {
+    qualification::check_pipeline(data, size);
+    return 0;
+}

scripts/generate-adversarial-corpus.py

@@ -0,0 +1,59 @@
+"""Deterministic synthetic seeds. Never consumes real logs or contacts a service."""
+import argparse
+import json
+import pathlib
+
+
+def generate(root):
+    entries = []
+    def save(target, name, content, purpose):
+        path = root / target / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(content)
+        entries.append({'path': path.relative_to(root).as_posix(), 'bytes': len(content),
+                        'provenance': 'authored synthetic fixture', 'purpose': purpose})
+    messages = [
+        ('sshd', 'Failed password for user-a from 203.0.113.1 port 50101 ssh2'),
+        ('sshd', 'Failed password for from from 203.0.113.1 port 50101 ssh2'),
+        ('sshd', 'Failed password for illegal user user-a from 2001:db8::1 port 50101 ssh2'),
+        ('sshd', 'Accepted keyboard-interactive/pam for user-a from 203.0.113.1 port 50101 ssh2'),
+        ('sshd', 'Failed publickey for user-a from 203.0.113.1 port 50101 ssh2'),
+        ('sshd', 'maximum authentication attempts exceeded for user-a from 203.0.113.1 port 50101 ssh2'),
+        ('sshd', 'Failed password for user-a from 203.0.113.1. port 50101 ssh2'),
+        ('sshd', 'Connection closed by 203.0.113.1 port 50101 [preauth]'),
+        ('sudo', 'user-a : TTY=pts/0 ; PWD=/ ; USER=root ; COMMAND=/usr/bin/echo from example'),
+        ('sudo', 'user-a : TTY=pts/0 ; PWD=/ ; USER=root ; COMMAND=/usr/bin/echo incorrect password attempt'),
+        ('su', 'FAILED SU (to root) user-a on pts/0'),
+        ('pam_unix(sshd:auth)', 'authentication failure; rhost=203.0.113.1 user=user-a'),
+    ]
+    for mode, prefix in [('syslog', 'Mar 10 09:00:00 '), ('journal', 'Tue 2026-03-10 17:00:00 +0800 ')]:
+        for index, (program, message) in enumerate(messages):
+            line = f'{prefix}example-host {program}[4200]: {message}'.encode()
+            save('parser', f'{mode}-{index:02}', line, 'supported/unsupported/auth-field boundaries')
+    seed = b'Mar 10 09:00:00 example-host sshd: Failed password for user-a from 203.0.113.1'
+    for host in ('example-host', 'example.example'):
+        for tag in ('sshd', 'sshd[0]', 'sshd[2147483647]', 'sshd[2147483648]'):
+            save('parser', 'envelope-' + host + '-' + tag,
+                 f'Mar 10 09:00:00 {host} {tag}: Failed password for user-a from 203.0.113.1 port 50101 ssh2'.encode(),
+                 'hostname and PID presence/int boundary')
+    for name, content in [('truncated', seed[:45]), ('nul', seed + b'\x00'), ('invalid-utf8', seed.replace(b'example-host', b'example-\xff'))]:
+        save('parser', name, content, 'byte/truncation boundaries')
+    for day in ('0', '1', '31', '32', '-255', '257', '2147483648'):
+        save('timestamp_day', 'day-' + day.replace('-', 'negative-'), day.encode(), 'integer/day domain')
+    for name, family, users in [('ssh-threshold', 0, [0] * 5), ('sudo-threshold', 11, [0] * 3), ('multi-user', 0, [0, 1, 2])]:
+        seed = bytearray([0])
+        for user in users:
+            seed.extend([family, 0, user, 0])
+        save('pipeline', name, bytes(seed), 'authored threshold witness; 4-byte records')
+    seed = bytearray([1])
+    for index in range(10):
+        seed.extend([0, 0 if index < 5 else 11, 0, 0])
+    save('pipeline', 'separated-episodes', bytes(seed), 'two five-event bursts separated by 1802 seconds')
+    (root / 'manifest.json').write_text(json.dumps({'generator_version': 1, 'entries': entries}, indent=2) + '\n')
+    print(f'{len(entries)} synthetic seeds written')
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--output', type=pathlib.Path, required=True)
+    generate(parser.parse_args().output)

scripts/measure-adversarial-performance.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""Matched local CLI observations; synthetic inputs, no historical timing claims."""
+import argparse
+import json
+from pathlib import Path
+import subprocess
+import time
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--baseline', type=Path, required=True)
+    parser.add_argument('--candidate', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    parser.add_argument('--lines', type=int, default=3000)
+    workloads = ('dense', 'unsupported', 'long-fields', 'distinct-subjects', 'window-boundary', 'mixed')
+    parser.add_argument('--workload', choices=workloads, help='rerun only the affected workload')
+    args = parser.parse_args()
+    args.output.mkdir(parents=True, exist_ok=False)
+    results = []
+    for workload in (args.workload,) if args.workload else workloads:
+        rows = []
+        for i in range(args.lines):
+            seconds = (i % 2) * 600 if workload == 'window-boundary' else i % 600
+            user = 'user-' + str(i % 7)
+            # Keep the original 65,024-subject period, inside the documentation prefix.
+            address = f'2001:db8::{i % (254 * 256) + 1:x}' if workload == 'distinct-subjects' else '203.0.113.1'
+            program = 'sshd'
+            message = f'Failed password for {user} from {address} port 50101 ssh2'
+            if workload == 'unsupported' or (workload == 'mixed' and i % 3 == 0):
+                message = 'unsupported synthetic message'
+            if workload == 'long-fields':
+                program = 'sudo'
+                message = 'user-a : TTY=pts/0 ; PWD=/ ; USER=root ; COMMAND=/usr/bin/echo ' + 'x' * 4096
+            if workload == 'mixed' and i % 3 == 1:
+                message = message.replace('Failed', 'Accepted')
+            rows.append(f'Mar 10 09:{seconds // 60:02}:{seconds % 60:02} example-host {program}[4200]: {message}\n')
+        source = args.output / (workload + '.log')
+        source.write_text(''.join(rows), encoding='utf-8')
+        for repeat in range(3):
+            # Alternate execution order to reduce a systematic warm-cache advantage.
+            builds = [('baseline', args.baseline), ('candidate', args.candidate)]
+            if repeat % 2:
+                builds.reverse()
+            for label, binary in builds:
+                target = args.output / f'{workload}-{repeat}-{label}'
+                metrics = args.output / f'{workload}-{repeat}-{label}.time'
+                command = ['/usr/bin/time', '-f', '%e %M', '-o', str(metrics), str(binary.resolve()),
+                           '--mode', 'syslog', '--year', '2026', '--csv', str(source.resolve()), str(target.resolve())]
+                started = time.monotonic()
+                run = subprocess.run(command, capture_output=True, timeout=60)
+                (args.output / f'{workload}-{repeat}-{label}.stderr').write_bytes(run.stderr)
+                observation = dict(workload=workload, repeat=repeat, build=label, command=command,
+                                   exit=run.returncode, wall_seconds=time.monotonic() - started,
+                                   lines=args.lines, input_bytes=source.stat().st_size)
+                if metrics.exists():
+                    observation['time_elapsed_rss_kib'] = metrics.read_text().strip()
+                results.append(observation)
+                (args.output / 'results.json').write_text(json.dumps(results, indent=2) + '\n')
+                if run.returncode:
+                    raise SystemExit(run.returncode)
+        print(workload, 'complete', flush=True)
+
+
+if __name__ == '__main__':
+    main()

scripts/run-adversarial-qualification.py

@@ -0,0 +1,69 @@
+"""Local sanitizer campaign runner with retained commands, exits and raw logs."""
+import argparse
+import json
+import os
+import pathlib
+import shutil
+import subprocess
+import sys
+import time
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--evidence', type=pathlib.Path, required=True)
+    parser.add_argument('--seconds', type=int, default=60)
+    parser.add_argument('--seed', type=int, default=129130)
+    args = parser.parse_args()
+    if not 1 <= args.seconds <= 900:
+        parser.error('seconds must be 1..900 per target')
+    source = pathlib.Path(__file__).resolve().parents[1]
+    evidence = args.evidence.resolve()
+    evidence.mkdir(parents=True, exist_ok=True)
+    manifest_path = evidence / 'execution.json'
+    if manifest_path.exists():
+        parser.error('use a new evidence directory; existing runs are immutable')
+    build = evidence / 'build'
+    records = []
+    env = dict(os.environ, ASAN_OPTIONS='detect_leaks=1', UBSAN_OPTIONS='halt_on_error=1:print_stacktrace=1')
+    def run(label, command, timeout):
+        start = time.monotonic()
+        with (evidence / (label + '.log')).open('wb') as log:
+            try:
+                result = subprocess.run([str(x) for x in command], stdout=log, stderr=subprocess.STDOUT,
+                                        env=env, timeout=timeout, cwd=source)
+                code = result.returncode
+            except subprocess.TimeoutExpired:
+                code = 'TIMEOUT'
+        records.append({'label': label, 'command': [str(x) for x in command], 'exit': code,
+                        'seconds': time.monotonic() - start,
+                        'ASAN_OPTIONS': env['ASAN_OPTIONS'], 'UBSAN_OPTIONS': env['UBSAN_OPTIONS']})
+        manifest_path.write_text(json.dumps(records, indent=2) + '\n')
+        print(label, code, flush=True)
+        if code != 0:
+            raise SystemExit(1)
+    run('configure', ['cmake', '-S', source, '-B', build, '-G', 'Ninja',
+        '-DCMAKE_CXX_COMPILER=clang++', '-DCMAKE_BUILD_TYPE=RelWithDebInfo',
+        '-DLOGLENS_WARNINGS_AS_ERRORS=ON', '-DLOGLENS_BUILD_FUZZERS=ON',
+        '-DCMAKE_CXX_FLAGS=-fsanitize=address,undefined -fno-omit-frame-pointer',
+        '-DCMAKE_EXE_LINKER_FLAGS=-fsanitize=address,undefined'], 120)
+    run('build', ['cmake', '--build', build, '-j2'], 600)
+    run('ctest', ['ctest', '--test-dir', build, '--output-on-failure'], 180)
+    corpus = evidence / 'corpus'
+    run('generate', [sys.executable, source / 'scripts/generate-adversarial-corpus.py', '--output', corpus], 30)
+    for target, folder, max_len in [('fuzz_parser', 'parser', 4096), ('fuzz_timestamp_day', 'timestamp_day', 32), ('fuzz_pipeline', 'pipeline', 97)]:
+        existing = source / 'tests/fuzz/corpus' / folder
+        if existing.is_dir():
+            for seed in existing.iterdir():
+                if seed.is_file():
+                    shutil.copyfile(seed, corpus / folder / ('existing-' + seed.name))
+        artifacts = evidence / 'failures' / folder
+        artifacts.mkdir(parents=True, exist_ok=True)
+        run(target, [build / target, corpus / folder, f'-seed={args.seed}',
+            f'-max_total_time={args.seconds}', f'-max_len={max_len}', '-timeout=3',
+            '-rss_limit_mb=1024', '-print_final_stats=1',
+            '-artifact_prefix=' + str(artifacts) + os.sep], args.seconds + 60)
+
+
+if __name__ == '__main__':
+    main()

src/parser.cpp

@@ -85,10 +85,19 @@ std::optional<Event> AuthLogParser::parse_line(std::string_view line,
 
     auto result = parser_internal::parse_source_envelope(config_, line, line_number);
     if (result.matched && result.event.has_value()) {
-        if (auto source_ip_failure = parser_internal::classify_source_ip_failure(*result.event)) {
-            result = std::move(*source_ip_failure);
-        } else {
-            result = parser_internal::dispatch_program(*result.event);
+        const auto source = std::move(*result.event);
+        result = parser_internal::dispatch_program(source);
+        // Validate the field actually emitted by the selected handler. Opaque
+        // usernames and command arguments must not be scanned as IP metadata.
+        if (result.event.has_value()) {
+            if (!result.event->source_ip.empty()) {
+                if (auto failure = parser_internal::classify_source_ip_failure(*result.event)) {
+                    result = std::move(*failure);
+                }
+            }
+        } else if (auto failure = parser_internal::classify_source_ip_failure(source)) {
+            // Preserve malformed-source telemetry for unsupported messages.
+            result = std::move(*failure);
         }
     }
 

src/parser/failure_classifier.cpp

@@ -8,13 +8,21 @@ namespace loglens::parser_internal {
 namespace {
 
 bool is_valid_ipv4_token(std::string_view token) {
+    if (token.empty() || token.back() == '.') {
+        return false;
+    }
     int parts = 0;
     while (!token.empty()) {
         const auto dot = token.find('.');
         const auto part = dot == std::string_view::npos ? token : token.substr(0, dot);
         if (part.empty()) {
             return false;
         }
+        for (const char ch : part) {
+            if (ch < '0' || ch > '9') {
+                return false;
+            }
+        }
 
         int value = 0;
         if (!parse_int(part, value) || value < 0 || value > 255) {
@@ -33,23 +41,55 @@ bool is_valid_ipv4_token(std::string_view token) {
 }
 
 bool is_valid_ipv6_like_token(std::string_view token) {
-    if (token.find(':') == std::string_view::npos) {
+    // Preserve the existing exclusion of the bare "::" token.
+    if (token == "::") {
         return false;
     }
-
-    bool saw_hex = false;
-    for (const char character : token) {
-        if (std::isxdigit(static_cast<unsigned char>(character)) != 0) {
-            saw_hex = true;
-            continue;
-        }
-        if (character == ':' || character == '.') {
-            continue;
-        }
+    const auto compression = token.find("::");
+    const bool compressed = compression != std::string_view::npos;
+    if (token.find(':') == std::string_view::npos
+        || (compressed && token.find("::", compression + 2) != std::string_view::npos)) {
         return false;
     }
-
-    return saw_hex;
+    unsigned groups = 0;
+    const auto count_side = [&groups](std::string_view side, bool allow_ipv4) {
+        while (!side.empty()) {
+            const auto colon = side.find(':');
+            const auto part = side.substr(0, colon);
+            if (part.empty()) {
+                return false;
+            }
+            if (part.find('.') != std::string_view::npos) {
+                if (!allow_ipv4 || colon != std::string_view::npos || !is_valid_ipv4_token(part)) {
+                    return false;
+                }
+                groups += 2;
+            } else {
+                if (part.size() > 4) {
+                    return false;
+                }
+                for (const unsigned char ch : part) {
+                    if (std::isxdigit(ch) == 0) {
+                        return false;
+                    }
+                }
+                ++groups;
+            }
+            if (colon == std::string_view::npos) {
+                return true;
+            }
+            side.remove_prefix(colon + 1);
+            if (side.empty()) {
+                return false;
+            }
+        }
+        return true;
+    };
+    if (!compressed) {
+        return count_side(token, true) && groups == 8;
+    }
+    return count_side(token.substr(0, compression), false)
+        && count_side(token.substr(compression + 2), true) && groups < 8;
 }
 
 bool is_valid_source_ip_token(std::string_view token) {
@@ -215,7 +255,7 @@ ParserFailureCategory failure_category_for_unrecognized_event(const Event& event
 }  // namespace
 
 std::optional<HandlerResult> classify_source_ip_failure(const Event& event) {
-    const auto candidate = extract_source_ip_candidate(event);
+    const auto candidate = event.source_ip.empty() ? extract_source_ip_candidate(event) : event.source_ip;
     if (candidate.empty() || is_valid_source_ip_token(candidate)) {
         return std::nullopt;
     }