Adversarial Robustness：攻击没成功，模型就鲁棒吗？

"""CPU protocol experiment. Binary linear model; no image/LLM benchmark."""
import json
import platform
from pathlib import Path
import numpy as np

OUT = Path(__file__).resolve().parent / 'results'
SEEDS = (79, 80, 81)
EPS = (0., .05, .1, .2, .3)
L2, LR, STEPS = .01, .2, 1000


def sigmoid(x):
    return np.exp(-np.logaddexp(0., -x))


def loss_grad(x, y, w, b):
    margin = y * (x @ w + b)
    a = -y * sigmoid(-margin)
    loss = np.logaddexp(0., -margin).mean() + L2 * (w @ w) / 2
    return loss, x.T @ a / len(y) + L2 * w, a.mean()


def input_grad(x, y, w, b):
    return (-y * sigmoid(-y * (x @ w + b)))[:, None] * w


def generate(rng, n):
    # These are abstract sensor features, not bounded pixels.
    y = rng.choice([-1., 1.], n)
    x = rng.normal(0., .3, (n, 20)) + .15 * y[:, None]
    x[:, 0] = y + rng.normal(0., .6, n)
    return x, y


def select_worse(x, candidate, y, w, b):
    mask = y * (candidate @ w + b) < y * (x @ w + b)
    return np.where(mask[:, None], candidate, x)


def pgd(x, y, w, b, eps, alpha, steps, restarts, rng, random_start):
    # Retain best per sample across all iterates, restarts AND the clean input.
    best = x.copy()
    for _ in range(restarts):
        z = x + rng.uniform(-eps, eps, x.shape) if random_start else x.copy()
        best = select_worse(best, z, y, w, b)
        for _ in range(steps):
            z = z + alpha * np.sign(input_grad(z, y, w, b))
            z = np.clip(z, x - eps, x + eps)
            best = select_worse(best, z, y, w, b)
    return best


def metrics(x, y, w, b, candidate, exact_margin):
    clean = y * (x @ w + b) > 0
    margin = y * (candidate @ w + b)
    assert np.min(np.abs(margin)) > 1e-10  # no ambiguous ties in this dataset
    survived = clean & (margin > 0)
    exact_ok = exact_margin > 0
    assert np.all(~exact_ok | survived)
    return dict(clean_accuracy=float(clean.mean()),
                robust_accuracy=float(survived.mean()),
                conditional_asr=float((clean & ~survived).sum() / clean.sum()),
                clean_correct=int(clean.sum()), failures=int((~survived).sum()),
                missed_failures=int((survived & ~exact_ok).sum()),
                max_margin_gap=float(np.max(margin - exact_margin)))


def gradient_check(x, y, w, b):
    _, gw, gb = loss_grad(x, y, w, b)
    h, errors = 1e-5, []
    for j in (0, 1, 7, 19):
        wp, wm = w.copy(), w.copy()
        wp[j] += h
        wm[j] -= h
        errors.append(abs((loss_grad(x, y, wp, b)[0] -
                           loss_grad(x, y, wm, b)[0]) / (2*h) - gw[j]))
    errors.append(abs((loss_grad(x, y, w, b+h)[0] -
                       loss_grad(x, y, w, b-h)[0]) / (2*h) - gb))
    gx = input_grad(x[:1], y[:1], w, b)[0]
    for j in (0, 1, 7, 19):
        xp, xm = x[:1].copy(), x[:1].copy()
        xp[0, j] += h
        xm[0, j] -= h
        lp = np.logaddexp(0., -y[0]*(xp[0] @ w + b))
        lm = np.logaddexp(0., -y[0]*(xm[0] @ w + b))
        errors.append(abs((lp-lm)/(2*h)-gx[j]))
    assert max(errors) < 1e-8
    return max(errors)


def main():
    OUT.mkdir(exist_ok=True)
    runs, checks = [], []
    for seed in SEEDS:
        rng = np.random.default_rng(seed)
        train, yt = generate(rng, 1024)
        dev, yd = generate(rng, 256)
        test, y = generate(rng, 2048)
        assert len(np.unique(np.vstack([train, dev, test]), axis=0)) == 3328
        w, b = rng.normal(0., .01, 20), 0.
        trace = []
        check = gradient_check(train[:32], yt[:32], w, b)
        for step in range(STEPS):
            loss, gw, gb = loss_grad(train, yt, w, b)
            if step % 100 == 0:
                trace.append(dict(step=step, objective=float(loss)))
            w -= LR * gw
            b -= LR * gb
        final_loss, gw, gb = loss_grad(train, yt, w, b)
        trace.append(dict(step=STEPS, objective=float(final_loss)))
        clean_margin = y * (test @ w + b)
        assert np.min(np.abs(clean_margin)) > 1e-10
        payload = dict(train=train, train_y=yt, dev=dev, dev_y=yd,
                       test=test, test_y=y, w=w, b=np.array(b))
        frozen = w.copy(), b
        row = dict(seed=seed, trace=trace, dev_accuracy=float((yd*(dev@w+b)>0).mean()),
                   final_gradient_norm=float(np.sqrt(gw@gw+gb*gb)), budgets=[])
        for k, eps in enumerate(EPS):
            attack_rng = np.random.default_rng(seed * 1000 + k)
            exact_margin = clean_margin - eps*np.abs(w).sum()
            assert np.min(np.abs(exact_margin)) > 1e-10
            random_best = test.copy()
            for _ in range(16):
                z = test + attack_rng.uniform(-eps, eps, test.shape)
                random_best = select_worse(random_best, z, y, w, b)
            candidates = dict(
                random16=random_best,
                pgd2=pgd(test, y, w, b, eps, eps/20, 2, 1, attack_rng, False),
                fgsm=test + eps*np.sign(input_grad(test, y, w, b)),
                pgd20=pgd(test, y, w, b, eps, eps/5, 20, 3, attack_rng, True),
                zero_gradient_fixture=test.copy())
            budget_row = dict(epsilon=eps, exact_robust_accuracy=float((exact_margin>0).mean()), attacks={})
            payload[f'e{k}_exact_margin'] = exact_margin
            union_survived = clean_margin > 0
            for name, z in candidates.items():
                assert np.max(np.abs(z-test)) <= eps + 1e-12
                assert np.isfinite(z).all()
                payload[f'e{k}_{name}'] = z
                budget_row['attacks'][name] = metrics(test, y, w, b, z, exact_margin)
                previous = union_survived.copy()
                union_survived &= y * (z @ w + b) > 0
                assert np.all(~union_survived | previous)
                if name in ('fgsm', 'pgd20'):
                    assert np.allclose(y*(z@w+b), exact_margin, atol=1e-10, rtol=0)
            budget_row['union_robust_accuracy'] = float(union_survived.mean())
            row['budgets'].append(budget_row)
        assert np.array_equal(w, frozen[0]) and b == frozen[1]
        exact_curve = [v['exact_robust_accuracy'] for v in row['budgets']]
        assert all(a >= b for a, b in zip(exact_curve, exact_curve[1:]))
        np.savez_compressed(OUT / f'seed{seed}.npz', **payload)
        checks.append(dict(seed=seed, gradient_max_error=check))
        runs.append(row)
        print('seed', seed, 'train', train.shape, 'test', test.shape, 'logits', (len(y),),
              'gradient_error', check, 'exact_curve', exact_curve)
    aggregate = []
    for k, eps in enumerate(EPS):
        ar = dict(epsilon=eps, attacks={})
        for name in runs[0]['budgets'][k]['attacks']:
            ar['attacks'][name] = {}
            for metric in ('clean_accuracy', 'robust_accuracy', 'conditional_asr', 'missed_failures'):
                values = [r['budgets'][k]['attacks'][name][metric] for r in runs]
                ar['attacks'][name][metric] = dict(mean=float(np.mean(values)), sd=float(np.std(values, ddof=1)))
        aggregate.append(ar)
    summary = dict(environment=dict(python=platform.python_version(), numpy=np.__version__, device='CPU', dtype='float64'),
                   configuration=dict(seeds=SEEDS, epsilon=EPS, l2=L2, lr=LR, steps=STEPS,
                   domain='R^20; fixed latent label; no pixel clipping; no feature scaling',
                   training='ordinary regularized logistic regression; final checkpoint; dev logging only'),
                   gradient_checks=checks, runs=runs, aggregate=aggregate)
    (OUT/'summary.json').write_text(json.dumps(summary, indent=2)+'\n')
    print(json.dumps(aggregate[3], indent=2))


if __name__ == '__main__':
    main()