# On Google Colab, install the scikit-learn version the course was built with.
# On your own computer, install requirements.txt instead; this cell then does nothing.
import subprocess, sys
if "google.colab" in sys.modules:
    subprocess.run([sys.executable, "-m", "pip", "install", "-q", "scikit-learn==1.9.0"], check=True)

Lab 7b. Evading a network detector under realistic constraints

This lab trains a differentiable detector on the course network flows and then attacks it with gradient steps, first with only an L-infinity bound in feature space and then under the kind of constraints a real attacker faces (Lesson 7.2). An image attack only has to stay small and visually imperceptible; a network attack has to keep the connection functional, so only some features can change, only in feasible directions, with integer counts staying integers and rates staying in range. Comparing the two settings shows that feature-space evasion rates overstate what an attacker can actually do. It runs in under a minute on a laptop CPU.

You will:

1. train a small differentiable detector on flows.csv numeric features;
2. select the attack records the detector catches;
3. run an unconstrained gradient attack in feature space;
4. run a constrained attack that respects attacker control, feasible direction, integrality, bounds, feature consistency and a cost budget;
5. report the evasion rate in both settings by attack family and discuss which constraints mattered.

The columns label, attack_type and ticket_opened are never used as features.

import os
import warnings

import numpy as np
import pandas as pd
import torch
import torch.nn as nn

warnings.filterwarnings("ignore", category=UserWarning)
torch.set_num_threads(4)
SEED = 20251015
torch.manual_seed(SEED)
rng = np.random.default_rng(SEED)

LOCAL = "../../data/flows.csv"
URL = "https://raw.githubusercontent.com/aalbata/machine-learning-for-cybersecurity/main/data/flows.csv"
flows = pd.read_csv(LOCAL if os.path.exists(LOCAL) else URL)
flows = flows.drop_duplicates()
print(flows.shape)

(36770, 21)

1. Features and a differentiable detector

We use twelve numeric features. Six of them are heavily skewed, so the model applies log1p to those before standardizing; the transform is part of the model, so gradients flow back to the raw feature values and the attack can work in the original units. The detector is a small multilayer perceptron, trained on the first ten days and tested on the last four, the time-ordered split of Unit 2.

SKEWED = ["duration", "src_bytes", "dst_bytes", "packets", "count", "srv_count"]
PLAIN = ["failed_logins", "serror_rate", "rerror_rate", "same_srv_rate", "dst_host_count", "hour"]
FEATURES = SKEWED + PLAIN
skew_mask = torch.tensor([f in SKEWED for f in FEATURES])

train = flows[flows["day"] <= 9]
test = flows[flows["day"] >= 10]
Xtr = torch.tensor(train[FEATURES].to_numpy(np.float32))
ytr = torch.tensor(train["label"].to_numpy(np.float32))
Xte = torch.tensor(test[FEATURES].to_numpy(np.float32))
yte = torch.tensor(test["label"].to_numpy(np.float32))


def transform(x):
    x = torch.where(skew_mask, torch.log1p(x.clamp(min=0)), x)
    return x


with torch.no_grad():
    Z = transform(Xtr)
    mean = Z.mean(0)
    std = Z.std(0) + 1e-6


class Detector(nn.Module):
    def __init__(self, d):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(d, 32), nn.ReLU(), nn.Linear(32, 1))

    def forward(self, x):
        z = (transform(x) - mean) / std
        return self.net(z).squeeze(-1)


model = Detector(len(FEATURES))
opt = torch.optim.Adam(model.parameters(), lr=5e-3, weight_decay=1e-4)
loss_fn = nn.BCEWithLogitsLoss(pos_weight=torch.tensor((ytr == 0).sum() / (ytr == 1).sum()))
for ep in range(60):
    model.train()
    opt.zero_grad()
    loss = loss_fn(model(Xtr), ytr)
    loss.backward()
    opt.step()
model.eval()


@torch.no_grad()
def scores(x):
    return torch.sigmoid(model(x))


THRESH = 0.5
test_scores = scores(Xte).numpy()
pred = (test_scores >= THRESH).astype(int)
tp = int(((pred == 1) & (yte.numpy() == 1)).sum())
fp = int(((pred == 1) & (yte.numpy() == 0)).sum())
fn = int(((pred == 0) & (yte.numpy() == 1)).sum())
recall = tp / (tp + fn)
precision = tp / (tp + fp) if tp + fp else 0.0
print(f"test recall {recall:.3f}, precision {precision:.3f}")

test recall 0.549, precision 0.397

2. The records to attack

An evasion attack starts from attack records the detector already flags. We collect the detected attacks in the test period and record their family.

test_attack_mask = (yte.numpy() == 1) & (pred == 1)
atk_idx = np.flatnonzero(test_attack_mask)
X_atk = Xte[atk_idx].clone()
fam_atk = test["attack_type"].to_numpy()[atk_idx]
print(f"detected attacks to evade: {len(atk_idx)}")
for f in sorted(set(fam_atk)):
    print(f"  {f}: {(fam_atk == f).sum()}")

detected attacks to evade: 237
  bruteforce: 71
  dos: 50
  exfil: 31
  scan: 85

3. An unconstrained feature-space attack

This attack moves every feature freely to lower the detector's score, bounded only by an L-infinity budget measured in standard deviations of each feature. It ignores whether the resulting record could exist. This is the setting that often appears in image-style evaluations.

feat_std_raw = Xtr.std(0) + 1e-6  # per-feature scale in raw units


def unconstrained_attack(X, eps=1.5, steps=50, alpha=None):
    if alpha is None:
        alpha = 2.5 * eps / steps
    X0 = X.clone()
    Xa = X.clone()
    for _ in range(steps):
        Xa.requires_grad_(True)
        s = model(Xa)  # logit; lower means less likely to alert
        grad, = torch.autograd.grad(s.sum(), Xa)
        Xa = Xa.detach() - alpha * feat_std_raw * grad.sign()
        # project to the L-infinity ball of radius eps standard deviations
        delta = torch.clamp((Xa - X0) / feat_std_raw, -eps, eps)
        Xa = (X0 + delta * feat_std_raw).detach()
    return Xa


X_unc = unconstrained_attack(X_atk)
unc_scores = scores(X_unc).numpy()
unc_evaded = unc_scores < THRESH
print(f"unconstrained evasion rate: {unc_evaded.mean():.3f}")

unconstrained evasion rate: 0.802

4. A constrained, problem-space-style attack

Now we impose the constraints an attacker on the network actually faces. The attack takes sign-gradient steps on the controllable features and enforces every constraint after each step, so every iterate is a feasible record. It runs once for each of five step sizes and keeps, for each record, the lowest-scoring iterate, which plays the role that restarts play for PGD.

• Attacker control. Only features the attacker shapes through its own connections may change. These are the duration, the bytes sent (source bytes), the bytes it causes the server to return (destination bytes, for example by requesting more data), the packet count, the connection count and the service count. Server-side outcomes (the error rates), failed logins, the destination host count and the hour are left fixed, and the same-service rate changes only through the consistency rule below.
• Feasible direction. Bytes, duration and packets can only grow, since the attacker can add traffic but not remove the traffic its own activity produces. Connection and service counts may grow or shrink, because an attacker can also pace its activity, for example a slower scan.
• Integrality. Packet, connection and service counts stay integers.
• Bounds and consistency. Durations and byte counts stay nonnegative, the packet and connection counts stay at least one, the service count stays between zero and the connection count, and the same-service rate is recomputed as the service count divided by the connection count, the relation that holds (to three decimals) for every record in flows.csv.
• Cost budget. The sum over the controllable features of the absolute change divided by one plus the original value (a relative change that stays defined at zero) is capped at 3.0, modeling a limit on how much the attacker will alter its traffic.

CONTROLLABLE = ["duration", "src_bytes", "dst_bytes", "packets", "count", "srv_count"]
INCREASE_ONLY = ["duration", "src_bytes", "dst_bytes", "packets"]
INTEGER = ["packets", "count", "srv_count"]
ctrl_mask = torch.tensor([f in CONTROLLABLE for f in FEATURES])
inc_mask = torch.tensor([f in INCREASE_ONLY for f in FEATURES])
fi = {f: i for i, f in enumerate(FEATURES)}


def enforce(Xa, X0, budget, direction=True):
    Xa = Xa.clone()
    # fixed features revert to their original values
    Xa[:, ~ctrl_mask] = X0[:, ~ctrl_mask]
    # feasible direction: byte, duration and packet features may only increase
    # (when relaxed, they stay nonnegative)
    if direction:
        Xa = torch.where(inc_mask, torch.maximum(Xa, X0), Xa)
    else:
        Xa = torch.where(inc_mask, Xa.clamp(min=0), Xa)
    # cost budget: cap the sum of |change| / (1 + |original|) over controllable features
    rel = (Xa - X0) / (X0.abs() + 1.0)
    total = (rel.abs() * ctrl_mask).sum(1, keepdim=True)
    scale = torch.clamp(budget / (total + 1e-9), max=1.0)
    Xa = X0 + (Xa - X0) * scale
    # integrality and bounds: durations and bytes at least 0 (kept by the direction step
    # and the budget scaling), packets and count at least 1, 0 <= srv_count <= count
    for f in INTEGER:
        Xa[:, fi[f]] = torch.round(Xa[:, fi[f]])
    Xa[:, fi["packets"]] = Xa[:, fi["packets"]].clamp(min=1)
    Xa[:, fi["count"]] = Xa[:, fi["count"]].clamp(min=1)
    Xa[:, fi["srv_count"]] = torch.minimum(Xa[:, fi["srv_count"]].clamp(min=0), Xa[:, fi["count"]])
    # consistency: same_srv_rate = srv_count / count, as in flows.csv
    Xa[:, fi["same_srv_rate"]] = (Xa[:, fi["srv_count"]] / Xa[:, fi["count"]]).clamp(0, 1)
    return Xa.detach()


def constrained_attack(X, steps=80, alphas=(0.02, 0.05, 0.1, 0.2, 0.5), budget=3.0, direction=True):
    """Projected sign-gradient descent on the detector's score, run once per step size.
    Every iterate satisfies the constraints; for each record the lowest-scoring iterate is kept."""
    X0 = X.clone()
    best = X.clone()
    best_s = scores(X).clone()
    for alpha in alphas:
        Xa = X.clone()
        for _ in range(steps):
            Xa.requires_grad_(True)
            s = model(Xa)
            grad, = torch.autograd.grad(s.sum(), Xa)
            step = alpha * feat_std_raw * grad.sign()
            Xa = Xa.detach() - step * ctrl_mask  # move only controllable features, toward lower score
            Xa = enforce(Xa, X0, budget, direction)
            s_new = scores(Xa)
            better = s_new < best_s
            best[better] = Xa[better]
            best_s[better] = s_new[better]
    return best


X_con = constrained_attack(X_atk)
con_scores = scores(X_con).numpy()
con_evaded = con_scores < THRESH
print(f"constrained evasion rate: {con_evaded.mean():.3f}")

constrained evasion rate: 0.118

5. Evasion by family, and which constraints mattered

We break the evasion rate down by attack family in both settings. Families whose detectable signal lives in features the attacker cannot change, or cannot change within the cost budget, resist the constrained attack even when the unconstrained attack evades them easily.

print(f"{'family':<12} {'n':>4} {'unconstrained':>14} {'constrained':>12}")
fam_rows = {}
for f in sorted(set(fam_atk)):
    m = fam_atk == f
    u = float(unc_evaded[m].mean())
    c = float(con_evaded[m].mean())
    fam_rows[f] = (int(m.sum()), u, c)
    print(f"{f:<12} {int(m.sum()):>4} {u:>14.3f} {c:>12.3f}")

# A quick look at which features the constrained attack actually moved, on average.
moved = (X_con - X_atk).abs().mean(0) / feat_std_raw
moved_ranked = sorted(zip(FEATURES, moved.tolist()), key=lambda kv: -kv[1])[:5]
print("most-changed features under constraints (in standard deviations):")
for name, v in moved_ranked:
    print(f"  {name}: {v:.3f}")

family          n  unconstrained  constrained
bruteforce     71          0.958        0.099
dos            50          0.720        0.280
exfil          31          1.000        0.129
scan           85          0.647        0.035
most-changed features under constraints (in standard deviations):
  srv_count: 2.505
  count: 2.187
  same_srv_rate: 0.140
  dst_bytes: 0.065
  duration: 0.000

The cell above shows that the unconstrained evasion rate is far higher than the constrained one, so a feature-space attack overstates the risk. To see which constraints matter, the next cell reruns the constrained attack twice more, each time relaxing one constraint and keeping the others. The first rerun lets byte, duration and packet features also decrease, and the second removes the cost budget.

ablation = {
    "all constraints": float(con_evaded.mean()),
    "features may also decrease": float((scores(constrained_attack(X_atk, direction=False)) < THRESH).float().mean()),
    "no cost budget": float((scores(constrained_attack(X_atk, budget=float("inf"))) < THRESH).float().mean()),
}
for name, v in ablation.items():
    print(f"{name:<34} evasion {v:.3f}")

all constraints                    evasion 0.118
features may also decrease         evasion 0.118
no cost budget                     evasion 0.527

Compare each relaxed setting with the first line. A constraint whose removal raises the evasion rate a lot is one the detector's margin depends on; a constraint whose removal changes little is not what stops the attack here.

Try this

• Loosen the cost budget from 3.0 to 10.0 and see how much the constrained evasion rate rises.
• Add the error-rate features to the controllable set (pretend the attacker can also suppress server errors) and see which families become evadable.
• Lower the detector's threshold to raise recall, then recompute both evasion rates. Predict first whether evasion becomes easier or harder when every attack must be pushed below a lower score.

Results summary

These are the numbers quoted in the lesson text.

results = {
    "n_detected_attacks": int(len(atk_idx)),
    "recall": float(recall),
    "precision": float(precision),
    "unconstrained_evasion": float(unc_evaded.mean()),
    "constrained_evasion": float(con_evaded.mean()),
    "evasion_ratio": float(unc_evaded.mean() / con_evaded.mean()) if con_evaded.mean() > 0 else float("inf"),
    "budget": 3.0,
    "eps_unconstrained": 1.5,
    "abl_no_direction": ablation["features may also decrease"],
    "abl_no_budget": ablation["no cost budget"],
}
for f in sorted(set(fam_atk)):
    n, u, c = fam_rows[f]
    results[f"unc_evasion_{f}"] = float(u)
    results[f"con_evasion_{f}"] = float(c)
for k, v in results.items():
    print(f"{k:<24} {v}")

export = {
    "families": sorted(set(fam_atk)),
    "n_by_family": [fam_rows[f][0] for f in sorted(set(fam_atk))],
    "unconstrained_by_family": [fam_rows[f][1] for f in sorted(set(fam_atk))],
    "constrained_by_family": [fam_rows[f][2] for f in sorted(set(fam_atk))],
    "overall_unconstrained": float(unc_evaded.mean()),
    "overall_constrained": float(con_evaded.mean()),
    "moved_features": [name for name, _ in moved_ranked],
    "moved_values": [v for _, v in moved_ranked],
}

if os.environ.get("COURSE_RESULTS"):
    import json
    with open(os.environ["COURSE_RESULTS"], "w") as f:
        json.dump({"results": results, "data": export}, f)

n_detected_attacks       237
recall                   0.5486111111111112
precision                0.3969849246231156
unconstrained_evasion    0.8016877637130801
constrained_evasion      0.11814345991561181
evasion_ratio            6.785714285714286
budget                   3.0
eps_unconstrained        1.5
abl_no_direction         0.1181434616446495
abl_no_budget            0.5274261832237244
unc_evasion_bruteforce   0.9577464788732394
con_evasion_bruteforce   0.09859154929577464
unc_evasion_dos          0.72
con_evasion_dos          0.28
unc_evasion_exfil        1.0
con_evasion_exfil        0.12903225806451613
unc_evasion_scan         0.6470588235294118
con_evasion_scan         0.03529411764705882