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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
14
TASKS.md
14
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 |
|
||||
|-----|--------|------|
|
||||
|
||||
6
TODO.md
6
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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -139,6 +139,7 @@ format = "text" # Log format: "text" (default) or "json"
|
||||
| `!archive <url>` | Save URL to Wayback Machine |
|
||||
| `!resolve <host> [host2 ...] [type]` | Bulk DNS resolution via TCP/SOCKS5 |
|
||||
| `!shorten <url>` | Shorten a URL via FlaskPaste |
|
||||
| `!paste <text>` | Create a paste via FlaskPaste |
|
||||
| `!pastemoni <add\|del\|list\|check>` | 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
|
||||
|
||||
@@ -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 <text>")
|
||||
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 <text>")
|
||||
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")
|
||||
|
||||
136
tests/test_cidr.py
Normal file
136
tests/test_cidr.py
Normal file
@@ -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]
|
||||
147
tests/test_defang.py
Normal file
147
tests/test_defang.py
Normal file
@@ -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]
|
||||
169
tests/test_dns_plugin.py
Normal file
169
tests/test_dns_plugin.py
Normal file
@@ -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]
|
||||
141
tests/test_encode.py
Normal file
141
tests/test_encode.py
Normal file
@@ -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()
|
||||
148
tests/test_hash.py
Normal file
148
tests/test_hash.py
Normal file
@@ -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]
|
||||
Reference in New Issue
Block a user