KeyWraith/tests/test_stream.py at main · Krishita17/KeyWraith · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import numpy as np

from keywraith import experiments as ex
from keywraith.stream import StreamDetector
from keywraith.synthetic import KeystrokeStreamGenerator, MachineProfile
from keywraith.attacker import AdaptiveAttacker


def _bundle():
    return ex.train(n_human=300, n_machine=300, seed=17)


def test_stream_flags_burst_injection_quickly():
    bundle = _bundle()
    gen = KeystrokeStreamGenerator(seed=50)
    burst = gen.machine_session(200, MachineProfile(mode="burst", device_delay_ms=5.0))
    sd = StreamDetector(bundle.detector, window=40, step=5, consecutive=3)
    res = sd.scan_session(burst)
    assert res["sustained_alert"] is True
    assert res["detect_at_keystroke"] is not None
    assert res["detect_at_keystroke"] <= 120  # well before the session ends


def test_stream_does_not_flag_human():
    bundle = _bundle()
    gen = KeystrokeStreamGenerator(seed=51)
    human = gen.human_session(300)
    sd = StreamDetector(bundle.detector, window=40, step=5, consecutive=3)
    res = sd.scan_session(human)
    assert res["sustained_alert"] is False


def test_push_and_scan_are_equivalent():
    bundle = _bundle()
    gen = KeystrokeStreamGenerator(seed=52)
    sess = gen.machine_session(160, MachineProfile(mode="burst", device_delay_ms=5.0))

    sd_batch = StreamDetector(bundle.detector, window=40, step=5, consecutive=3)
    batch = sd_batch.scan_session(sess)

    sd_online = StreamDetector(bundle.detector, window=40, step=5, consecutive=3)
    online_scores = []
    for ev in sess.events:
        v = sd_online.push(ev)
        if v is not None:
            online_scores.append(round(v.score, 6))
    batch_scores = [round(v.score, 6) for v in batch["verdicts"]]
    assert online_scores == batch_scores
    assert sd_online.alerted == batch["sustained_alert"]


def test_window_too_small_rejected():
    bundle = _bundle()
    try:
        StreamDetector(bundle.detector, window=3)
        assert False, "expected ValueError"
    except ValueError:
        pass


def test_stream_evaluation_summary():
    bundle = _bundle()
    res = ex.stream_evaluation(bundle, seed=60, n=30)
    assert res["injection_catch_rate"] > 0.8
    assert res["human_false_alarm_rate"] < 0.2