test_detector.py

import numpy as np

from keywraith import experiments as ex
from keywraith.detector import BehavioralDetector
from keywraith.baseline import SpeedThresholdBaseline
from keywraith.alert import build_alert, render_alert
from keywraith.attacker import AdaptiveAttacker
from keywraith.synthetic import KeystrokeStreamGenerator


def _bundle():
    # Small, fast dataset for tests.
    return ex.train(n_human=300, n_machine=300, seed=17)


def test_detector_beats_baseline_and_is_strong():
    bundle = _bundle()
    m = ex.evaluate(bundle)
    assert m["detector"]["roc_auc"] > 0.95
    assert m["detector"]["roc_auc"] >= m["baseline"]["roc_auc"] - 1e-9
    assert m["detector"]["false_positive_rate"] < 0.1


def test_save_load_roundtrip(tmp_path):
    bundle = _bundle()
    path = tmp_path / "det.pkl"
    bundle.detector.save(str(path))
    loaded = BehavioralDetector.load(str(path))
    a = bundle.detector.predict_proba(bundle.X_test)
    b = loaded.predict_proba(bundle.X_test)
    assert np.allclose(a, b)


def test_alert_flags_injection_and_clears_human():
    bundle = _bundle()
    gen = KeystrokeStreamGenerator(seed=99)
    atk = AdaptiveAttacker(seed=100)

    injected = atk.humanized_session(200, humanization_level=0.5)
    human = gen.human_session(200)

    inj_alert = build_alert(injected, bundle.detector)
    hum_alert = build_alert(human, bundle.detector)

    assert inj_alert.verdict == "injected"
    assert hum_alert.verdict == "human"
    assert isinstance(render_alert(inj_alert), str)
    # Explanation direction is oriented, not just unsigned magnitude.
    assert any(f["direction"] == "pushes-injected" for f in inj_alert.top_features)


def test_arms_race_monotone_pressure_on_baseline():
    bundle = _bundle()
    arms = ex.arms_race(bundle, levels=[0.0, 0.4, 0.8], n_per_level=150, seed=7)
    base_rates = [r["baseline_detection_rate"] for r in arms["rows"]]
    # Humanizing (slowing down) should not make the speed baseline *better*.
    assert base_rates[0] >= base_rates[-1]
    # And the attacker pays a growing time cost to do it.
    costs = [r["attacker_slowdown_x"] for r in arms["rows"]]
    assert costs[-1] > costs[0]


def test_gradient_boosting_model_trains_and_scores():
    bundle = ex.train(n_human=250, n_machine=250, model="gradient_boosting", seed=17)
    m = ex.evaluate(bundle)
    assert m["detector"]["roc_auc"] > 0.9


def test_adversarial_retraining_recovers_high_humanization():
    adv = ex.adversarial_retrain(
        levels=[0.0, 0.6, 0.8], n_per_level=120, seed=17
    )
    idx = adv["levels"].index(0.8)
    std = adv["standard_detection_rate"][idx]
    aware = adv["evasion_aware_detection_rate"][idx]
    # Evasion-aware retraining reclaims detection the standard detector lost...
    assert aware > std
    # ...without wrecking human false positives.
    assert adv["evasion_aware_fpr"] < 0.1


def test_baseline_fit_to_fpr():
    gen = KeystrokeStreamGenerator(seed=5)
    humans = [gen.human_session(200) for _ in range(200)]
    base = SpeedThresholdBaseline().fit_to_fpr(humans, target_fpr=0.01)
    flagged = np.mean([base.predict_session(s) for s in humans])
    assert flagged <= 0.05  # keeps human false positives low