Files
DECNET/tests/correlation/test_bounty_dedup.py
anti f2b3393669 chore: relicense to AGPL-3.0-or-later and add SPDX headers
Replaces LICENSE (GPLv3 -> AGPLv3) and prepends
`SPDX-License-Identifier: AGPL-3.0-or-later` to every source file
across decnet/, decnet_web/, tests/, scripts/, and tools/.

Rationale: closes the GPLv3 ASP loophole so any party operating a
modified DECNET as a network service must offer their modified
source. Personal copyright (Samuel Paschuan) + inbound=outbound
contributions make a future unilateral relicense infeasible.

- LICENSE: full AGPL-3.0 text (gnu.org/licenses/agpl-3.0.txt)
- COPYRIGHT: project copyright notice
- tools/add_spdx_headers.py: idempotent header injector
  (shebang- and PEP 263-aware)

Touches 1565 source files (.py, .ts, .tsx, .js, .jsx, .css, .sh).
No behavior change; comments only.
2026-05-22 21:04:16 -04:00

86 lines
2.5 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-or-later
"""
Tests for bounty deduplication.
Identical (bounty_type, attacker_ip, payload) tuples must be dropped so
aggressive scanners cannot saturate the bounty table.
"""
import pytest
from decnet.web.db.factory import get_repository
@pytest.fixture
async def repo(tmp_path):
r = get_repository(db_path=str(tmp_path / "test.db"))
await r.initialize()
return r
_BASE = {
"decky": "decky-01",
"service": "ssh",
"attacker_ip": "10.0.0.1",
"bounty_type": "credential",
"payload": {"username": "admin", "password": "password"},
}
@pytest.mark.anyio
async def test_duplicate_dropped(repo):
await repo.add_bounty({**_BASE})
await repo.add_bounty({**_BASE})
bounties = await repo.get_bounties()
assert len(bounties) == 1
@pytest.mark.anyio
async def test_different_ip_not_deduped(repo):
await repo.add_bounty({**_BASE})
await repo.add_bounty({**_BASE, "attacker_ip": "10.0.0.2"})
bounties = await repo.get_bounties()
assert len(bounties) == 2
@pytest.mark.anyio
async def test_different_type_not_deduped(repo):
await repo.add_bounty({**_BASE})
await repo.add_bounty({**_BASE, "bounty_type": "fingerprint"})
bounties = await repo.get_bounties()
assert len(bounties) == 2
@pytest.mark.anyio
async def test_different_payload_not_deduped(repo):
await repo.add_bounty({**_BASE})
await repo.add_bounty({**_BASE, "payload": {"username": "root", "password": "toor"}})
bounties = await repo.get_bounties()
assert len(bounties) == 2
@pytest.mark.anyio
async def test_flood_protection(repo):
for _ in range(50):
await repo.add_bounty({**_BASE})
bounties = await repo.get_bounties()
assert len(bounties) == 1
@pytest.mark.anyio
async def test_dict_payload_dedup(repo):
"""Payload passed as dict (pre-serialisation path) is still deduped."""
await repo.add_bounty({**_BASE, "payload": {"username": "admin", "password": "password"}})
await repo.add_bounty({**_BASE, "payload": {"username": "admin", "password": "password"}})
bounties = await repo.get_bounties()
assert len(bounties) == 1
@pytest.mark.anyio
async def test_string_payload_dedup(repo):
"""Payload passed as pre-serialised string is also deduped."""
import json
p = json.dumps({"username": "admin", "password": "password"})
await repo.add_bounty({**_BASE, "payload": p})
await repo.add_bounty({**_BASE, "payload": p})
bounties = await repo.get_bounties()
assert len(bounties) == 1