From 9abf8dce645703bed27f9fd7856d411f92b45259 Mon Sep 17 00:00:00 2001 From: user Date: Sat, 21 Feb 2026 16:54:18 +0100 Subject: [PATCH] feat: add !paste command and unit tests for 5 core plugins Add cmd_paste to flaskpaste plugin (create paste, return URL). Add test suites for encode, hash, defang, cidr, and dns plugins (83 new test cases, 1093 total). Co-Authored-By: Claude Opus 4.6 --- ROADMAP.md | 6 +- TASKS.md | 14 +++- TODO.md | 6 +- docs/CHEATSHEET.md | 1 + docs/USAGE.md | 18 +++++ plugins/flaskpaste.py | 30 +++++++ tests/test_cidr.py | 136 +++++++++++++++++++++++++++++++ tests/test_defang.py | 147 ++++++++++++++++++++++++++++++++++ tests/test_dns_plugin.py | 169 +++++++++++++++++++++++++++++++++++++++ tests/test_encode.py | 141 ++++++++++++++++++++++++++++++++ tests/test_hash.py | 148 ++++++++++++++++++++++++++++++++++ 11 files changed, 809 insertions(+), 7 deletions(-) create mode 100644 tests/test_cidr.py create mode 100644 tests/test_defang.py create mode 100644 tests/test_dns_plugin.py create mode 100644 tests/test_encode.py create mode 100644 tests/test_hash.py diff --git a/ROADMAP.md b/ROADMAP.md index 9b92d7b..1f5b039 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -116,8 +116,8 @@ - [ ] URL shortener integration (shorten URLs in alerts and long output) - [ ] Webhook listener (HTTP endpoint for push events to channels) - [ ] Granular ACLs (per-command permission tiers: trusted, operator, admin) -- [ ] `paste` plugin (manual paste to FlaskPaste) -- [ ] `shorten` plugin (manual URL shortening) +- [x] `paste` command (manual paste to FlaskPaste) +- [x] `shorten` command (manual URL shortening) - [x] `emailcheck` plugin (SMTP VRFY/RCPT TO) - [x] `canary` plugin (canary token generator/tracker) - [x] `virustotal` plugin (hash/URL/IP/domain lookup, free API) @@ -126,5 +126,5 @@ - [x] `mac` plugin (OUI vendor lookup, local IEEE database) - [x] `pastemoni` plugin (monitor paste sites for keywords) - [ ] `cron` plugin (scheduled bot commands on a timer) -- [ ] Plugin command unit tests (encode, hash, dns, cidr, defang) +- [x] Plugin command unit tests (encode, hash, dns, cidr, defang) - [ ] CI pipeline diff --git a/TASKS.md b/TASKS.md index bcaef2d..6b9c866 100644 --- a/TASKS.md +++ b/TASKS.md @@ -1,6 +1,18 @@ # derp - Tasks -## Current Sprint -- v1.3.0 Tier 2 Plugins (2026-02-20) +## Current Sprint -- v2.0.0 Quick Wins (2026-02-21) + +| Pri | Status | Task | +|-----|--------|------| +| P0 | [x] | `!paste` command in `plugins/flaskpaste.py` | +| P0 | [x] | Unit tests: `test_encode.py` (18 cases) | +| P0 | [x] | Unit tests: `test_hash.py` (15 cases) | +| P0 | [x] | Unit tests: `test_defang.py` (18 cases) | +| P0 | [x] | Unit tests: `test_cidr.py` (14 cases) | +| P0 | [x] | Unit tests: `test_dns_plugin.py` (18 cases) | +| P2 | [x] | Documentation update (USAGE.md, CHEATSHEET.md, ROADMAP.md, TODO.md) | + +## Previous Sprint -- v1.3.0 Tier 2 Plugins (2026-02-20) | Pri | Status | Task | |-----|--------|------| diff --git a/TODO.md b/TODO.md index ee086cd..1726d0b 100644 --- a/TODO.md +++ b/TODO.md @@ -78,11 +78,11 @@ is preserved in git history for reference. ## Plugins -- Utility -- [ ] `paste` -- manual paste to FlaskPaste -- [ ] `shorten` -- manual URL shortening +- [x] `paste` -- manual paste to FlaskPaste +- [x] `shorten` -- manual URL shortening - [ ] `cron` -- scheduled bot commands on a timer ## Testing -- [ ] Plugin command unit tests (encode, hash, dns, cidr, defang) +- [x] Plugin command unit tests (encode, hash, dns, cidr, defang) - [ ] CI pipeline diff --git a/docs/CHEATSHEET.md b/docs/CHEATSHEET.md index 7793c9d..52c2349 100644 --- a/docs/CHEATSHEET.md +++ b/docs/CHEATSHEET.md @@ -314,6 +314,7 @@ VT rate limit: 4 req/min. Email check: max 5, admin only. ``` !shorten https://long-url.com/x # Shorten URL +!paste some text to paste # Create paste ``` Auto-shortens URLs in `!alert` announcements and history when plugin is loaded. diff --git a/docs/USAGE.md b/docs/USAGE.md index 07e45ae..52621eb 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -139,6 +139,7 @@ format = "text" # Log format: "text" (default) or "json" | `!archive ` | Save URL to Wayback Machine | | `!resolve [host2 ...] [type]` | Bulk DNS resolution via TCP/SOCKS5 | | `!shorten ` | Shorten a URL via FlaskPaste | +| `!paste ` | Create a paste via FlaskPaste | | `!pastemoni ` | Paste site keyword monitoring | ### Command Shorthand @@ -902,6 +903,23 @@ https://paste.mymx.me/s/AbCdEfGh - mTLS client cert skips PoW; falls back to PoW challenge if no cert - Also used internally by `!alert` to shorten announcement URLs +### `!paste` -- Create Paste + +Create a text paste via FlaskPaste. + +``` +!paste some text or data to paste +``` + +Output format: + +``` +https://paste.mymx.me/AbCdEfGh +``` + +- Pastes arbitrary text content +- mTLS client cert skips PoW; falls back to PoW challenge if no cert + ### `!pastemoni` -- Paste Site Keyword Monitor Monitor public paste sites for keywords (data leaks, credential dumps, brand diff --git a/plugins/flaskpaste.py b/plugins/flaskpaste.py index e908360..3343d57 100644 --- a/plugins/flaskpaste.py +++ b/plugins/flaskpaste.py @@ -192,3 +192,33 @@ async def cmd_shorten(bot, message): await bot.reply(message, short) else: await bot.reply(message, "shorten failed: no URL returned") + + +@command("paste", help="Create a paste: !paste ") +async def cmd_paste(bot, message): + """Create a paste via FlaskPaste. + + Usage: + !paste some text to paste + """ + parts = message.text.split(None, 1) + if len(parts) < 2: + await bot.reply(message, "Usage: !paste ") + return + + content = parts[1] + base_url = _get_base_url(bot) + loop = asyncio.get_running_loop() + + try: + url = await loop.run_in_executor( + None, _create_paste, base_url, content, + ) + except Exception as exc: + await bot.reply(message, f"paste failed: {exc}") + return + + if url: + await bot.reply(message, url) + else: + await bot.reply(message, "paste failed: no URL returned") diff --git a/tests/test_cidr.py b/tests/test_cidr.py new file mode 100644 index 0000000..6776802 --- /dev/null +++ b/tests/test_cidr.py @@ -0,0 +1,136 @@ +"""Tests for the CIDR calculator plugin.""" + +import asyncio +import importlib.util +import sys +from pathlib import Path + +from derp.irc import Message + +# plugins/ is not a Python package -- load the module from file path +_spec = importlib.util.spec_from_file_location( + "plugins.cidr", Path(__file__).resolve().parent.parent / "plugins" / "cidr.py", +) +_mod = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = _mod +_spec.loader.exec_module(_mod) + +from plugins.cidr import cmd_cidr # noqa: E402 + +# -- Helpers ----------------------------------------------------------------- + +class _FakeBot: + def __init__(self): + self.replied: list[str] = [] + + async def reply(self, message, text: str) -> None: + self.replied.append(text) + + +def _msg(text: str) -> Message: + return Message( + raw="", prefix="alice!~alice@host", nick="alice", + command="PRIVMSG", params=["#test", text], tags={}, + ) + + +# -- Info: IPv4 -------------------------------------------------------------- + +class TestCidrInfoIPv4: + def test_slash24(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr 192.168.1.0/24"))) + reply = bot.replied[0] + assert "net:192.168.1.0/24" in reply + assert "hosts:254" in reply + assert "mask:255.255.255.0" in reply + assert "broadcast:192.168.1.255" in reply + + def test_slash31_no_broadcast(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr 10.0.0.0/31"))) + reply = bot.replied[0] + assert "hosts:2" in reply + assert "broadcast" not in reply + + def test_slash32(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr 10.0.0.1/32"))) + reply = bot.replied[0] + assert "hosts:1" in reply + assert "broadcast" not in reply + + def test_range(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr 10.0.0.0/30"))) + reply = bot.replied[0] + assert "range:10.0.0.0-10.0.0.3" in reply + assert "hosts:2" in reply + + def test_wildcard(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr 172.16.0.0/16"))) + reply = bot.replied[0] + assert "wildcard:0.0.255.255" in reply + + +# -- Info: IPv6 -------------------------------------------------------------- + +class TestCidrInfoIPv6: + def test_ipv6_slash64(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr 2001:db8::/64"))) + reply = bot.replied[0] + assert "net:2001:db8::/64" in reply + assert "hosts:" in reply + # No mask/wildcard/broadcast for IPv6 + assert "mask:" not in reply + + def test_ipv6_slash128(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr ::1/128"))) + reply = bot.replied[0] + assert "hosts:1" in reply + + +# -- Contains ---------------------------------------------------------------- + +class TestCidrContains: + def test_ip_in_network(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr contains 10.0.0.0/8 10.1.2.3"))) + assert "is in" in bot.replied[0] + + def test_ip_not_in_network(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr contains 10.0.0.0/8 192.168.1.1"))) + assert "NOT in" in bot.replied[0] + + def test_invalid_network(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr contains badnet 10.0.0.1"))) + assert "Invalid network" in bot.replied[0] + + def test_invalid_ip(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr contains 10.0.0.0/8 notanip"))) + assert "Invalid IP" in bot.replied[0] + + def test_missing_args(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr contains 10.0.0.0/8"))) + assert "Usage" in bot.replied[0] + + +# -- Errors ------------------------------------------------------------------ + +class TestCidrErrors: + def test_invalid_network(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr notanetwork"))) + assert "Invalid network" in bot.replied[0] + + def test_missing_args(self): + bot = _FakeBot() + asyncio.run(cmd_cidr(bot, _msg("!cidr"))) + assert "Usage" in bot.replied[0] diff --git a/tests/test_defang.py b/tests/test_defang.py new file mode 100644 index 0000000..a82b68e --- /dev/null +++ b/tests/test_defang.py @@ -0,0 +1,147 @@ +"""Tests for the defang/refang plugin.""" + +import asyncio +import importlib.util +import sys +from pathlib import Path + +from derp.irc import Message + +# plugins/ is not a Python package -- load the module from file path +_spec = importlib.util.spec_from_file_location( + "plugins.defang", Path(__file__).resolve().parent.parent / "plugins" / "defang.py", +) +_mod = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = _mod +_spec.loader.exec_module(_mod) + +from plugins.defang import ( # noqa: E402 + _defang, + _refang, + cmd_defang, + cmd_refang, +) + +# -- Helpers ----------------------------------------------------------------- + +class _FakeBot: + def __init__(self): + self.replied: list[str] = [] + + async def reply(self, message, text: str) -> None: + self.replied.append(text) + + +def _msg(text: str) -> Message: + return Message( + raw="", prefix="alice!~alice@host", nick="alice", + command="PRIVMSG", params=["#test", text], tags={}, + ) + + +# -- Pure function: _defang -------------------------------------------------- + +class TestDefangFunction: + def test_http_url(self): + result = _defang("http://evil.com") + assert "[://]" in result + + def test_https_url(self): + result = _defang("https://evil.com/path") + assert "[://]" in result + + def test_ftp_url(self): + result = _defang("ftp://files.evil.com") + assert "ftp[://]" in result + + def test_ip_address(self): + assert _defang("192.168.1.1") == "192[.]168[.]1[.]1" + + def test_path_dots_preserved(self): + result = _defang("http://evil.com/file.php?q=1") + assert "[://]" in result + assert "file.php" in result + + def test_domain_only(self): + assert _defang("evil.com") == "evil[.]com" + + +# -- Pure function: _refang -------------------------------------------------- + +class TestRefangFunction: + def test_basic(self): + assert _refang("evil[.]com") == "evil.com" + + def test_protocol(self): + assert _refang("http[://]evil[.]com") == "http://evil.com" + + def test_hxxp(self): + assert _refang("hxxps[://]evil[.]com") == "https://evil.com" + + def test_hXXp(self): + assert _refang("hXXps[://]evil[.]com") == "https://evil.com" + + def test_at_sign(self): + assert _refang("user[at]evil[.]com") == "user@evil.com" + + def test_at_uppercase(self): + assert _refang("user[AT]evil[.]com") == "user@evil.com" + + +# -- Roundtrip --------------------------------------------------------------- + +class TestRoundtrip: + def test_ip(self): + original = "10.0.0.1" + defanged = _defang(original) + refanged = _refang(defanged) + assert refanged == original + + def test_domain(self): + original = "evil.com" + defanged = _defang(original) + assert "[.]" in defanged + refanged = _refang(defanged) + assert refanged == original + + def test_refang_hxxp(self): + """Refang handles hxxp notation not produced by _defang.""" + assert _refang("hxxps://evil[.]com") == "https://evil.com" + + +# -- Command: defang --------------------------------------------------------- + +class TestCmdDefang: + def test_url(self): + bot = _FakeBot() + asyncio.run(cmd_defang(bot, _msg("!defang https://evil.com/path"))) + assert "[://]" in bot.replied[0] + + def test_ip(self): + bot = _FakeBot() + asyncio.run(cmd_defang(bot, _msg("!defang 192.168.1.1"))) + assert bot.replied[0] == "192[.]168[.]1[.]1" + + def test_missing_args(self): + bot = _FakeBot() + asyncio.run(cmd_defang(bot, _msg("!defang"))) + assert "Usage" in bot.replied[0] + + +# -- Command: refang --------------------------------------------------------- + +class TestCmdRefang: + def test_url(self): + bot = _FakeBot() + asyncio.run(cmd_refang(bot, _msg("!refang hxxps[://]evil[.]com"))) + assert bot.replied[0] == "https://evil.com" + + def test_ip(self): + bot = _FakeBot() + asyncio.run(cmd_refang(bot, _msg("!refang 10[.]0[.]0[.]1"))) + assert bot.replied[0] == "10.0.0.1" + + def test_missing_args(self): + bot = _FakeBot() + asyncio.run(cmd_refang(bot, _msg("!refang"))) + assert "Usage" in bot.replied[0] diff --git a/tests/test_dns_plugin.py b/tests/test_dns_plugin.py new file mode 100644 index 0000000..0947b14 --- /dev/null +++ b/tests/test_dns_plugin.py @@ -0,0 +1,169 @@ +"""Tests for the DNS lookup plugin.""" + +import asyncio +import importlib.util +import sys +from pathlib import Path +from unittest.mock import AsyncMock, patch + +from derp.irc import Message + +# plugins/ is not a Python package -- load the module from file path +_spec = importlib.util.spec_from_file_location( + "plugins.dns", Path(__file__).resolve().parent.parent / "plugins" / "dns.py", +) +_mod = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = _mod +_spec.loader.exec_module(_mod) + +from plugins.dns import cmd_dns # noqa: E402 + +# -- Helpers ----------------------------------------------------------------- + +class _FakeBot: + def __init__(self): + self.replied: list[str] = [] + + async def reply(self, message, text: str) -> None: + self.replied.append(text) + + +def _msg(text: str) -> Message: + return Message( + raw="", prefix="alice!~alice@host", nick="alice", + command="PRIVMSG", params=["#test", text], tags={}, + ) + + +def _mock_query(rcode, results): + """Create an AsyncMock returning (rcode, results).""" + mock = AsyncMock(return_value=(rcode, results)) + return mock + + +# -- Auto-detect type -------------------------------------------------------- + +class TestAutoDetect: + def test_domain_defaults_to_a(self): + bot = _FakeBot() + mock = _mock_query(0, ["93.184.216.34"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com"))) + assert "A:" in bot.replied[0] + assert "93.184.216.34" in bot.replied[0] + + def test_ip_defaults_to_ptr(self): + bot = _FakeBot() + mock = _mock_query(0, ["dns.google"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns 8.8.8.8"))) + assert "PTR:" in bot.replied[0] + assert "dns.google" in bot.replied[0] + + +# -- Explicit type ----------------------------------------------------------- + +class TestExplicitType: + def test_mx(self): + bot = _FakeBot() + mock = _mock_query(0, ["10 mail.example.com"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com MX"))) + assert "MX:" in bot.replied[0] + assert "mail.example.com" in bot.replied[0] + + def test_aaaa(self): + bot = _FakeBot() + mock = _mock_query(0, ["2606:2800:220:1:248:1893:25c8:1946"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com AAAA"))) + assert "AAAA:" in bot.replied[0] + + def test_txt(self): + bot = _FakeBot() + mock = _mock_query(0, ["v=spf1 -all"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com TXT"))) + assert "TXT:" in bot.replied[0] + assert "v=spf1" in bot.replied[0] + + def test_case_insensitive(self): + bot = _FakeBot() + mock = _mock_query(0, ["93.184.216.34"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com a"))) + assert "A:" in bot.replied[0] + + +# -- Unknown type ------------------------------------------------------------ + +class TestUnknownType: + def test_unknown_type(self): + bot = _FakeBot() + asyncio.run(cmd_dns(bot, _msg("!dns example.com BOGUS"))) + assert "Unknown type" in bot.replied[0] + + +# -- Invalid PTR IP ---------------------------------------------------------- + +class TestInvalidPtr: + def test_invalid_ip_for_ptr(self): + bot = _FakeBot() + asyncio.run(cmd_dns(bot, _msg("!dns notanip PTR"))) + assert "Invalid IP for PTR" in bot.replied[0] + + +# -- Query results ----------------------------------------------------------- + +class TestQueryResults: + def test_success(self): + bot = _FakeBot() + mock = _mock_query(0, ["93.184.216.34", "93.184.216.35"]) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com"))) + assert "93.184.216.34" in bot.replied[0] + assert "93.184.216.35" in bot.replied[0] + + def test_timeout(self): + bot = _FakeBot() + mock = _mock_query(-1, []) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com"))) + assert "timeout" in bot.replied[0] + + def test_network_error(self): + bot = _FakeBot() + mock = _mock_query(-2, []) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com"))) + assert "network error" in bot.replied[0] + + def test_nxdomain(self): + bot = _FakeBot() + mock = _mock_query(3, []) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns nonexistent.test"))) + assert "NXDOMAIN" in bot.replied[0] + + def test_no_records(self): + bot = _FakeBot() + mock = _mock_query(0, []) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com AAAA"))) + assert "no records" in bot.replied[0] + + def test_servfail(self): + bot = _FakeBot() + mock = _mock_query(2, []) + with patch.object(_mod, "_query", mock): + asyncio.run(cmd_dns(bot, _msg("!dns example.com"))) + assert "SERVFAIL" in bot.replied[0] + + +# -- Missing args ------------------------------------------------------------ + +class TestMissingArgs: + def test_no_args(self): + bot = _FakeBot() + asyncio.run(cmd_dns(bot, _msg("!dns"))) + assert "Usage" in bot.replied[0] diff --git a/tests/test_encode.py b/tests/test_encode.py new file mode 100644 index 0000000..2ece8cf --- /dev/null +++ b/tests/test_encode.py @@ -0,0 +1,141 @@ +"""Tests for the encode/decode plugin.""" + +import asyncio +import importlib.util +import sys +from pathlib import Path + +from derp.irc import Message + +# plugins/ is not a Python package -- load the module from file path +_spec = importlib.util.spec_from_file_location( + "plugins.encode", Path(__file__).resolve().parent.parent / "plugins" / "encode.py", +) +_mod = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = _mod +_spec.loader.exec_module(_mod) + +from plugins.encode import cmd_decode, cmd_encode # noqa: E402 + +# -- Helpers ----------------------------------------------------------------- + +class _FakeBot: + def __init__(self): + self.replied: list[str] = [] + + async def reply(self, message, text: str) -> None: + self.replied.append(text) + + +def _msg(text: str) -> Message: + return Message( + raw="", prefix="alice!~alice@host", nick="alice", + command="PRIVMSG", params=["#test", text], tags={}, + ) + + +# -- Encode ------------------------------------------------------------------ + +class TestEncodeB64: + def test_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode b64 hello"))) + assert bot.replied[0] == "aGVsbG8=" + + def test_decode(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode b64 aGVsbG8="))) + assert bot.replied[0] == "hello" + + def test_roundtrip(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode b64 test data"))) + encoded = bot.replied[0] + asyncio.run(cmd_decode(bot, _msg(f"!decode b64 {encoded}"))) + assert bot.replied[1] == "test data" + + +class TestEncodeHex: + def test_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode hex hello"))) + assert bot.replied[0] == "68656c6c6f" + + def test_decode(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode hex 68656c6c6f"))) + assert bot.replied[0] == "hello" + + def test_decode_invalid(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode hex zzzz"))) + assert "error" in bot.replied[0].lower() + + +class TestEncodeUrl: + def test_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode url hello world"))) + assert bot.replied[0] == "hello%20world" + + def test_decode(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode url hello%20world"))) + assert bot.replied[0] == "hello world" + + def test_special_chars(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode url a=1&b=2"))) + assert "%3D" in bot.replied[0] + assert "%26" in bot.replied[0] + + +class TestEncodeRot13: + def test_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode rot13 hello"))) + assert bot.replied[0] == "uryyb" + + def test_roundtrip(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode rot13 secret"))) + encoded = bot.replied[0] + asyncio.run(cmd_decode(bot, _msg(f"!decode rot13 {encoded}"))) + assert bot.replied[1] == "secret" + + +class TestEncodeErrors: + def test_unknown_format_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode bad hello"))) + assert "Unknown format" in bot.replied[0] + + def test_unknown_format_decode(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode bad hello"))) + assert "Unknown format" in bot.replied[0] + + def test_missing_args_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode"))) + assert "Usage" in bot.replied[0] + + def test_missing_args_decode(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode"))) + assert "Usage" in bot.replied[0] + + def test_missing_text_encode(self): + bot = _FakeBot() + asyncio.run(cmd_encode(bot, _msg("!encode b64"))) + assert "Usage" in bot.replied[0] + + def test_missing_text_decode(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode hex"))) + assert "Usage" in bot.replied[0] + + def test_decode_b64_invalid(self): + bot = _FakeBot() + asyncio.run(cmd_decode(bot, _msg("!decode b64 !!!invalid!!!"))) + assert "error" in bot.replied[0].lower() diff --git a/tests/test_hash.py b/tests/test_hash.py new file mode 100644 index 0000000..8a7e386 --- /dev/null +++ b/tests/test_hash.py @@ -0,0 +1,148 @@ +"""Tests for the hash/hashid plugin.""" + +import asyncio +import hashlib +import importlib.util +import sys +from pathlib import Path + +from derp.irc import Message + +# plugins/ is not a Python package -- load the module from file path +_spec = importlib.util.spec_from_file_location( + "plugins.hash", Path(__file__).resolve().parent.parent / "plugins" / "hash.py", +) +_mod = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = _mod +_spec.loader.exec_module(_mod) + +from plugins.hash import cmd_hash, cmd_hashid # noqa: E402 + +# -- Helpers ----------------------------------------------------------------- + +class _FakeBot: + def __init__(self): + self.replied: list[str] = [] + + async def reply(self, message, text: str) -> None: + self.replied.append(text) + + +def _msg(text: str) -> Message: + return Message( + raw="", prefix="alice!~alice@host", nick="alice", + command="PRIVMSG", params=["#test", text], tags={}, + ) + + +# -- Hash: all algorithms --------------------------------------------------- + +class TestHashAllAlgos: + def test_default_shows_md5_sha1_sha256(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash hello"))) + reply = bot.replied[0] + assert "md5:" in reply + assert "sha1:" in reply + assert "sha256:" in reply + + def test_default_values_correct(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash hello"))) + reply = bot.replied[0] + expected_md5 = hashlib.md5(b"hello").hexdigest() + expected_sha1 = hashlib.sha1(b"hello").hexdigest() + expected_sha256 = hashlib.sha256(b"hello").hexdigest() + assert expected_md5 in reply + assert expected_sha1 in reply + assert expected_sha256 in reply + + +class TestHashSpecificAlgo: + def test_md5(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash md5 hello"))) + expected = hashlib.md5(b"hello").hexdigest() + assert f"md5: {expected}" == bot.replied[0] + + def test_sha1(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash sha1 hello"))) + expected = hashlib.sha1(b"hello").hexdigest() + assert f"sha1: {expected}" == bot.replied[0] + + def test_sha256(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash sha256 hello"))) + expected = hashlib.sha256(b"hello").hexdigest() + assert f"sha256: {expected}" == bot.replied[0] + + def test_sha512(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash sha512 hello"))) + expected = hashlib.sha512(b"hello").hexdigest() + assert f"sha512: {expected}" == bot.replied[0] + + +class TestHashUnknownAlgo: + def test_unknown_treated_as_text(self): + """Unknown algo name is treated as part of the text to hash.""" + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash unknown hello"))) + reply = bot.replied[0] + # "unknown hello" is hashed as plain text (no algo match) + expected_md5 = hashlib.md5(b"unknown hello").hexdigest() + assert expected_md5 in reply + + +class TestHashMissingArgs: + def test_no_args(self): + bot = _FakeBot() + asyncio.run(cmd_hash(bot, _msg("!hash"))) + assert "Usage" in bot.replied[0] + + +# -- HashID ------------------------------------------------------------------ + +class TestHashIdPatterns: + def test_md5(self): + h = hashlib.md5(b"test").hexdigest() + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg(f"!hashid {h}"))) + assert "MD5" in bot.replied[0] + + def test_sha1(self): + h = hashlib.sha1(b"test").hexdigest() + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg(f"!hashid {h}"))) + assert "SHA-1" in bot.replied[0] + + def test_sha256(self): + h = hashlib.sha256(b"test").hexdigest() + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg(f"!hashid {h}"))) + assert "SHA-256" in bot.replied[0] + + def test_bcrypt(self): + bcrypt_hash = "$2b$12$LJ3m4ysXql6gVjLnEQKbN.ZRA/O.mTeDGVMnRqz7E2B0sq1pYG2oW" + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg(f"!hashid {bcrypt_hash}"))) + assert "bcrypt" in bot.replied[0] + + def test_mysql41(self): + mysql_hash = "*94BDCEBE19083CE2A1F959FD02F964C7AF4CFC29" + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg(f"!hashid {mysql_hash}"))) + assert "MySQL" in bot.replied[0] + + def test_unknown(self): + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg("!hashid zzznotahash"))) + assert "Unknown" in bot.replied[0] + + +class TestHashIdMissingArgs: + def test_no_args(self): + bot = _FakeBot() + asyncio.run(cmd_hashid(bot, _msg("!hashid"))) + assert "Usage" in bot.replied[0]