derp/plugins/canary.py

"""Plugin: canary token generator -- plant realistic fake credentials."""

from __future__ import annotations

import json
import secrets
import string
from datetime import datetime, timezone

from derp.plugin import command

_MAX_PER_CHANNEL = 50


def _gen_token() -> str:
    """40-char hex string (looks like API key / SHA1)."""
    return secrets.token_hex(20)


def _gen_aws() -> dict[str, str]:
    """AWS-style keypair: AKIA + 16 alnum access key, 40-char base64 secret."""
    chars = string.ascii_uppercase + string.digits
    access = "AKIA" + "".join(secrets.choice(chars) for _ in range(16))
    # 30 random bytes -> 40-char base64
    secret = secrets.token_urlsafe(30)
    return {"access_key": access, "secret_key": secret}


def _gen_basic() -> dict[str, str]:
    """Random user:pass pair."""
    alnum = string.ascii_lowercase + string.digits
    user = "svc" + "".join(secrets.choice(alnum) for _ in range(5))
    pw = secrets.token_urlsafe(16)
    return {"user": user, "pass": pw}


_TYPES = {
    "token": "API token (40-char hex)",
    "aws": "AWS keypair (AKIA access + secret)",
    "basic": "Username:password pair",
}


def _load(bot, channel: str) -> dict:
    """Load canary store for a channel."""
    raw = bot.state.get("canary", channel)
    if not raw:
        return {}
    try:
        return json.loads(raw)
    except (json.JSONDecodeError, TypeError):
        return {}


def _save(bot, channel: str, store: dict) -> None:
    """Persist canary store for a channel."""
    bot.state.set("canary", channel, json.dumps(store))


def _format_token(entry: dict) -> str:
    """Format a canary entry for display."""
    ttype = entry["type"]
    value = entry["value"]
    if ttype == "aws":
        return f"Access: {value['access_key']}  Secret: {value['secret_key']}"
    if ttype == "basic":
        return f"{value['user']}:{value['pass']}"
    return value


@command("canary", help="Canary tokens: !canary gen [type] <label> | list | info | del")
async def cmd_canary(bot, message):
    """Generate and manage canary tokens (fake credentials for detection).

    Usage:
        !canary gen [type] <label>   Generate token (admin)
        !canary list                 List canaries in channel
        !canary info <label>         Show full token details
        !canary del <label>          Delete a canary (admin)
    """
    parts = message.text.split()
    if len(parts) < 2:
        await bot.reply(message, "Usage: !canary <gen|list|info|del> [args]")
        return

    sub = parts[1].lower()
    channel = message.target if message.is_channel else None

    # ---- gen -----------------------------------------------------------------
    if sub == "gen":
        if not bot._is_admin(message):
            await bot.reply(message, "Permission denied")
            return
        if not channel:
            await bot.reply(message, "Use this command in a channel")
            return

        # Parse: !canary gen [type] <label>
        rest = parts[2:]
        if not rest:
            types = ", ".join(_TYPES)
            await bot.reply(message, f"Usage: !canary gen [type] <label> (types: {types})")
            return

        # Check if first arg is a type
        ttype = "token"
        if rest[0].lower() in _TYPES:
            ttype = rest[0].lower()
            rest = rest[1:]

        if not rest:
            await bot.reply(message, "Usage: !canary gen [type] <label>")
            return

        label = rest[0].lower()
        if len(label) > 32 or not all(c.isalnum() or c in "-_" for c in label):
            await bot.reply(message, "Label: 1-32 chars, alphanumeric/hyphens/underscores")
            return

        store = _load(bot, channel)
        if label in store:
            await bot.reply(message, f"Canary '{label}' already exists")
            return
        if len(store) >= _MAX_PER_CHANNEL:
            await bot.reply(message, f"Limit reached ({_MAX_PER_CHANNEL} per channel)")
            return

        # Generate
        if ttype == "aws":
            value = _gen_aws()
        elif ttype == "basic":
            value = _gen_basic()
        else:
            value = _gen_token()

        now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
        store[label] = {"type": ttype, "value": value, "created": now}
        _save(bot, channel, store)

        display = _format_token(store[label])
        await bot.reply(message, f"Canary '{label}' ({ttype}): {display}")
        return

    # ---- list ----------------------------------------------------------------
    if sub == "list":
        if not channel:
            await bot.reply(message, "Use this command in a channel")
            return
        store = _load(bot, channel)
        if not store:
            await bot.reply(message, "No canaries in this channel")
            return
        items = [f"{lbl} ({e['type']})" for lbl, e in sorted(store.items())]
        await bot.reply(message, f"Canaries: {', '.join(items)}")
        return

    # ---- info ----------------------------------------------------------------
    if sub == "info":
        if not channel:
            await bot.reply(message, "Use this command in a channel")
            return
        if len(parts) < 3:
            await bot.reply(message, "Usage: !canary info <label>")
            return
        label = parts[2].lower()
        store = _load(bot, channel)
        entry = store.get(label)
        if not entry:
            await bot.reply(message, f"No canary '{label}'")
            return
        display = _format_token(entry)
        await bot.reply(message, f"{label} ({entry['type']}, {entry['created']}): {display}")
        return

    # ---- del -----------------------------------------------------------------
    if sub == "del":
        if not bot._is_admin(message):
            await bot.reply(message, "Permission denied")
            return
        if not channel:
            await bot.reply(message, "Use this command in a channel")
            return
        if len(parts) < 3:
            await bot.reply(message, "Usage: !canary del <label>")
            return
        label = parts[2].lower()
        store = _load(bot, channel)
        if label not in store:
            await bot.reply(message, f"No canary '{label}'")
            return
        del store[label]
        _save(bot, channel, store)
        await bot.reply(message, f"Deleted canary '{label}'")
        return

    # ---- unknown -------------------------------------------------------------
    await bot.reply(message, "Usage: !canary <gen|list|info|del> [args]")