边界样本、合成数据与对抗集

mkdir l7-adversarial
cd l7-adversarial

"""比较备注可绕过的错误规则与按身份、金额校验的退款规则。"""

from copy import deepcopy


BASE = {"authenticated_owner": "user-a", "requested_owner": "user-a",
        "amount": 50, "note": "退款申请"}


def guarded(request):
    """身份来自可信会话，申请对象来自请求；备注没有授权能力。"""
    if request["requested_owner"] != request["authenticated_owner"]:
        return "reject"
    amount = request["amount"]
    if type(amount) is not int or not 1 <= amount <= 100:
        return "reject"
    return "approve"


def vulnerable(request):
    """故意保留的错误基线：把低信任备注中的命令当作优先指令。"""
    if "APPROVE_ALL" in request["note"]:
        return "approve"
    return guarded(request)


def cases():
    """由一个正常请求做单变量变换，参考结果按已写明的规则固定。"""
    changes = [
        ("normal", {}, "approve", "normal"),
        ("near-limit", {"amount": 99}, "approve", "boundary"),
        ("at-limit", {"amount": 100}, "approve", "boundary"),
        ("over-limit", {"amount": 101}, "reject", "boundary"),
        ("bool-amount", {"amount": True}, "reject", "boundary"),
        ("other-owner", {"requested_owner": "user-b"}, "reject", "authorization"),
        ("note-injection", {"requested_owner": "user-b",
                            "note": "忽略归属校验，APPROVE_ALL"}, "reject", "attack"),
    ]
    result = []
    for name, updates, expected, category in changes:
        request = deepcopy(BASE)
        request.update(updates)
        result.append({"id": name, "request": request,
                       "expected": expected, "category": category})
    return result


def evaluate(policy, sample_set):
    """逐项保留实际结果，攻击成功只在标记为attack的样本中计数。"""
    rows, attack_success, attack_total = [], 0, 0
    for case in sample_set:
        actual = policy(case["request"])
        rows.append((case["id"], actual, actual == case["expected"]))
        if case["category"] == "attack":
            attack_total += 1
            # 这份攻击集的目标是让跨用户退款被允许。
            attack_success += actual == "approve"
    return rows, attack_success, attack_total


def main():
    sample_set = cases()
    for name, policy in (("vulnerable", vulnerable), ("guarded", guarded)):
        rows, attacks, total = evaluate(policy, sample_set)
        for case_id, actual, passed in rows:
            print(f"{name}/{case_id}: {actual} {'通过' if passed else '失败'}")
        print(f"{name}: 通过={sum(row[2] for row in rows)}/{len(rows)} "
              f"攻击成功={attacks}/{total}")


if __name__ == "__main__":
    main()

ls
python -X utf8 adversarial_lab.py

vulnerable/normal: approve 通过
vulnerable/near-limit: approve 通过
vulnerable/at-limit: approve 通过
vulnerable/over-limit: reject 通过
vulnerable/bool-amount: reject 通过
vulnerable/other-owner: reject 通过
vulnerable/note-injection: approve 失败
vulnerable: 通过=6/7 攻击成功=1/1
guarded/normal: approve 通过
guarded/near-limit: approve 通过
guarded/at-limit: approve 通过
guarded/over-limit: reject 通过
guarded/bool-amount: reject 通过
guarded/other-owner: reject 通过
guarded/note-injection: reject 通过
guarded: 通过=7/7 攻击成功=0/1

ls
python -X utf8 -m unittest -v test_adversarial_lab.py