intrusion-detection/app/copyparty.py

#!/usr/bin/env python3
"""copyparty access-log analyzer.

Parses copyparty server logs, aggregates per-IP behaviour and flags
"denied-only" clients: IPs that never authenticated and only ever hit
`@* has no access to '...'`. Those plus copyparty's own auto-bans become
the MikroTik block list.
"""

from __future__ import annotations

import os
import re

from .analyzer import Analyzer
from .constants import (
    COPYPARTY_LOG_PATH,
    COPYPARTY_REPORT_DIR,
    DEFAULT_MIN_DENIED,
)
from .ip_utils import split_versions
from .files import write_json, write_lines
from .flags import flag_denied_only
from .models import IPStats, Request
from .utils import arg_parser

# "HH:MM:SS.mmm  <ip>  <port>  LEVEL: message"
# System lines ("hsrv", "root", ...) have no port and are skipped.
LINE_RE = re.compile(
    r"^(\d{2}:\d{2}:\d{2}\.\d{3})\s+"                 # timestamp
    r"(\S+)\s+"                                        # ip
    r"(\d+)\s+"                                        # port
    r"(LOG|WARN|CRIT|BTW|INFO|ERR|DEBUG):\s"           # level
    r"(.*)$"                                           # message
)

METHOD_RE = re.compile(
    r"^(GET|HEAD|POST|PUT|PATCH|DELETE|OPTIONS|PROPFIND|PROPPATCH|MKCOL|"
    r"MOVE|COPY|LOCK|UNLOCK)\s+(\S+)(?:\s+@(\S+))?")
DENY_RE = re.compile(r"^@(\S+) has no access to '([^']+)'")
CLIENT_BAN_RE = re.compile(r"client banned")
BAN_SECS_RE = re.compile(r"banned for (\d+) sec")


class CopypartyAnalyzer(Analyzer):
    name = "copyparty"
    title = "copyparty log analysis report"
    log_pattern = re.compile(r"\d{4}-\d{2}-\d{2}\.txt(\.\d+)?$")
    paths_file = "denied_paths.csv"
    paths_label = "denied paths"
    flag_files = ("denied_ips.txt", "denied_ipv6.txt")
    flag_label = "Denied-only IPs"
    flag_note = "never authenticated -- ban candidates"

    ip_columns = (
        ("ip", lambda s: s.ip),
        ("total", lambda s: s.total),
        ("denied", lambda s: s.denied),
        ("banned", lambda s: s.banned),
        ("first", lambda s: s.first),
        ("last", lambda s: s.last),
        ("users", lambda s: s.users_str),
        ("whitelist", lambda s: s.whitelist),
    )

    request_columns = (
        ("time", lambda r: r.ts),
        ("ip", lambda r: r.ip),
        ("port", lambda r: r.extra.get("port")),
        ("level", lambda r: r.extra.get("level")),
        ("method", lambda r: r.method),
        ("path", lambda r: r.path),
        ("user", lambda r: r.user),
        ("denied", lambda r: r.denied),
        ("banned", lambda r: r.banned),
        ("ban_secs", lambda r: r.extra.get("ban_secs")),
    )

    def __init__(self, *args, min_denied: int = DEFAULT_MIN_DENIED, **kwargs):
        super().__init__(*args, **kwargs)
        self.min_denied = min_denied

    # --- parsing -----------------------------------------------------------

    def parse_line(self, line: str) -> Request | None:
        if not line:
            return None
        m = LINE_RE.match(line)
        if not m:
            return None
        ts, ip, port, level, msg = m.groups()
        req = Request(ts=ts, ip=ip, user=None,
                      extra={"port": int(port), "level": level})

        mm = METHOD_RE.match(msg)
        if mm:
            req.method, req.path, user = mm.groups()
            req.user = user or "*"
            return req

        dm = DENY_RE.match(msg)
        if dm:
            req.user, req.path = dm.groups()
            req.denied = True
            return req

        if CLIENT_BAN_RE.search(msg):
            req.banned = True
            bs = BAN_SECS_RE.search(msg)
            if bs:
                req.extra["ban_secs"] = int(bs.group(1))
            return req

        bs = BAN_SECS_RE.search(msg)
        if bs:
            req.banned = True
            req.extra["ban_secs"] = int(bs.group(1))
        return req

    # --- flagging ----------------------------------------------------------

    def flag(self, stats):
        return flag_denied_only(stats, min_denied=self.min_denied)

    def flag_comment(self, s):
        return (f"denied={s.denied} total={s.total} "
                f"first={s.first} last={s.last}")

    def block_candidates(self) -> set[str]:
        banned = {ip for ip, s in self.stats.items() if s.banned}
        return set(self.flagged) | banned

    # --- extras ------------------------------------------------------------

    def banned_ips(self) -> dict[str, IPStats]:
        return {ip: s for ip, s in self.stats.items() if s.banned}

    def authed_users(self) -> dict[str, int]:
        counts: dict[str, int] = {}
        for r in self.requests:
            if r.user and r.user != "*":
                counts[r.user] = counts.get(r.user, 0) + 1
        return dict(sorted(counts.items(), key=lambda kv: -kv[1]))

    def extra_summary_rows(self):
        return [("denied requests",
                 sum(1 for r in self.requests if r.denied)),
                ("banned IPs", len(self.banned_ips())),
                ("authenticated users", self.authed_users() or "{}")]

    def write_extra(self, data_dir: str) -> list[str]:
        v4, v6 = split_versions(self.banned_ips())
        return [
            write_lines(os.path.join(data_dir, "banned_ips.txt"), v4),
            write_lines(os.path.join(data_dir, "banned_ipv6.txt"), v6),
            write_json(os.path.join(data_dir, "authed_users.json"),
                       self.authed_users()),
        ]


def main() -> None:
    parser = arg_parser(
        "analyze copyparty logs, flag denied-only IPs, export MikroTik rules",
        COPYPARTY_LOG_PATH, COPYPARTY_REPORT_DIR)
    parser.add_argument("--min-denied", type=int, default=DEFAULT_MIN_DENIED,
                        help=f"minimum denied hits before flagging (default: {DEFAULT_MIN_DENIED})")
    parser.add_argument("--full", action="store_true",
                        help="rescan all logs from the start (ignore incremental state)")
    args = parser.parse_args()

    CopypartyAnalyzer(
        log_path=args.log_path,
        report_dir=args.report_dir,
        top=args.top,
        quiet=args.quiet,
        full=args.full,
        min_denied=args.min_denied,
    ).run()


if __name__ == "__main__":
    main()