commit c8b181ea47b1d4dc76d42a99155c212c42d91ad0 Author: dev6-ai Date: Sun Aug 23 15:41:17 2026 -0400 feat: initial upload of proxy-monitor diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..c7bf16a --- /dev/null +++ b/.dockerignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.git/ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b3260dd --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +data/ +__pycache__/ +*.log +.env diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..c17f482 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,10 @@ +FROM python:3.12-slim +RUN apt-get update && apt-get install -y --no-install-recommends \ + openssh-client && \ + rm -rf /var/lib/apt/lists/* +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY app/ app/ +EXPOSE 8080 +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"] diff --git a/Dockerfile.agent b/Dockerfile.agent new file mode 100644 index 0000000..89619e6 --- /dev/null +++ b/Dockerfile.agent @@ -0,0 +1,10 @@ +FROM python:3.12-slim +RUN apt-get update && apt-get install -y --no-install-recommends \ + iptables ipset && \ + rm -rf /var/lib/apt/lists/* +WORKDIR /app +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt +COPY app/ app/ +EXPOSE 8787 +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8787"] diff --git a/app/docker_ops.py b/app/docker_ops.py new file mode 100644 index 0000000..ac4673d --- /dev/null +++ b/app/docker_ops.py @@ -0,0 +1,124 @@ +import docker + +_client = None + + +def _c(): + global _client + if _client is None: + _client = docker.DockerClient(base_url="unix:///var/run/docker.sock", timeout=15) + return _client + + +def host_info(): + try: + import shutil + client = _c() + info = client.info() + containers = client.containers.list(all=True, sparse=True) + running = sum(1 for c in containers if c.status == "running") + images = len(client.images.list()) + mem_total = info.get("MemTotal", 0) + mem_used = 0 + try: + with open("/proc/meminfo") as f: + mem = {} + for line in f.read().splitlines(): + if ":" in line: + parts = line.split() + mem[parts[0].rstrip(":")] = int(parts[1]) * 1024 + mem_total = mem.get("MemTotal", mem_total) + mem_used = mem_total - mem.get("MemAvailable", mem.get("MemFree", 0)) + except Exception: + pass + disk_total = disk_used = 0 + try: + d = shutil.disk_usage("/") + disk_total, disk_used = d.total, d.used + except Exception: + pass + return { + "name": info.get("Name", "?"), + "os": info.get("OperatingSystem", "?"), + "version": info.get("ServerVersion", "?"), + "kernel": info.get("KernelVersion", "?"), + "cpu": info.get("NCPU", 0), + "memory": mem_total, + "mem_used": mem_used, + "mem_total": mem_total, + "running": running, + "containers": len(containers), + "images": images, + "disk_total": disk_total, + "disk_used": disk_used, + "storage": info.get("DriverStatus", [["?", "?"]])[0][1] if info.get("DriverStatus") else "?", + "error": None, + } + except Exception as e: + return {"name": "?", "running": 0, "containers": 0, "images": 0, "error": str(e)} + + +def list_containers(): + try: + client = _c() + return [ + { + "id": c.short_id, + "name": c.name, + "image": c.image.tags[0] if c.image.tags else c.image.short_id[:19], + "status": c.status, + "ports": c.ports, + "created": c.attrs.get("Created", "")[:19], + } + for c in client.containers.list(all=True) + ] + except Exception as e: + return {"error": str(e)} + + +def list_images(): + try: + client = _c() + return [{"id": i.short_id, "tags": i.tags, "size": i.attrs.get("Size", 0)} for i in client.images.list()] + except Exception as e: + return {"error": str(e)} + + +def container_action(cid, action): + try: + client = _c() + c = client.containers.get(cid) + if action == "start": + c.start() + elif action == "stop": + c.stop(timeout=10) + elif action == "restart": + c.restart(timeout=10) + else: + return {"error": f"unknown action {action}"} + return {"status": f"{action}ed", "name": c.name} + except Exception as e: + return {"error": str(e)} + + +def container_stats(cid): + try: + client = _c() + c = client.containers.get(cid) + s = c.stats(stream=False) + mem_usage = s["memory_stats"].get("usage", 0) + mem_limit = s["memory_stats"].get("limit", 1) + cpu_delta = s["cpu_stats"]["cpu_usage"]["total_usage"] - s["precpu_stats"]["cpu_usage"]["total_usage"] + system_delta = s["cpu_stats"]["system_cpu_usage"] - s["precpu_stats"]["system_cpu_usage"] + cpu_percent = (cpu_delta / system_delta) * 100.0 * s["cpu_stats"].get("online_cpus", 1) if system_delta > 0 else 0 + return { + "memory_usage": mem_usage, + "memory_limit": mem_limit, + "memory_pct": round(mem_usage / mem_limit * 100, 1) if mem_limit else 0, + "cpu_pct": round(cpu_percent, 1), + "pids": s["pids_stats"].get("current", 0), + "network_rx": s.get("networks", {}).get("eth0", {}).get("rx_bytes", 0), + "network_tx": s.get("networks", {}).get("eth0", {}).get("tx_bytes", 0), + } + except Exception as e: + return {"error": str(e)} diff --git a/app/firewall.py b/app/firewall.py new file mode 100644 index 0000000..8ef05c4 --- /dev/null +++ b/app/firewall.py @@ -0,0 +1,163 @@ +import json +import os +import subprocess +import time +from pathlib import Path + +SERVERS_FILE = Path(os.environ.get("SERVERS_FILE", "/cache/servers.json")) + +_fw_cache = {} + +DEFAULT_SERVERS = [ + {"name": "ubuntu-web (.65)", "host": "192.168.0.65", "user": "rob", "key": "/root/.ssh/id_ed25519", "sudo": True}, + {"name": "ubuntu VPS (.2)", "host": "66.179.94.204", "user": "rob", "key": "/root/.ssh/id_ed25519", "sudo": True}, + {"name": "jetson-nano (.23)", "host": "192.168.0.23", "user": "rob", "key": "/root/.ssh/id_ed25519", "sudo": True}, +] + + +def load_servers(): + if SERVERS_FILE.exists(): + try: + servers = json.loads(SERVERS_FILE.read_text()) + if servers: + return servers + except Exception: + pass + return DEFAULT_SERVERS + + +def ssh_exec(server, cmd, timeout=30): + # Local agent mode: run directly on this host (no SSH) + if server.get("host") == "local": + try: + r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout) + return r.returncode, r.stdout, r.stderr + except Exception as e: + return 1, "", str(e) + host = server.get("host") + user = server.get("user", "rob") + key = server.get("key", "/root/.ssh/id_ed25519") + ssh_cmd = [ + "ssh", "-i", key, + "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null", + "-o", "ConnectTimeout=10", + "-o", "BatchMode=yes", + f"{user}@{host}", cmd, + ] + try: + r = subprocess.run(ssh_cmd, capture_output=True, text=True, timeout=timeout) + return r.returncode, r.stdout, r.stderr + except Exception as e: + return 1, "", str(e) + + +def _sudo(server): + return "sudo " if server.get("sudo", True) else "" + + +def firewall_status(server): + key = f"{server.get('name')}:{server.get('host')}" + now = time.time() + item = _fw_cache.get(key) + if item and now - item[0] < 120: + return item[1] + s = _sudo(server) + status = { + "name": server.get("name", server.get("host")), + "host": server.get("host"), + "reachable": False, + "country_block": False, + "scanner_block": False, + "scanner_ips": 0, + "country_cidrs": 0, + "error": None, + } + + # single combined SSH call for all firewall checks + cmd = ( + "echo ok; " + f"{s}ipset list blocked_countries 2>/dev/null | grep -c /; " + f"{s}iptables -C INPUT -m set --match-set blocked_countries src -j DROP 2>/dev/null && echo cb_active || echo cb_inactive; " + f"{s}iptables -C INPUT -j SCANNER-BLOCK 2>/dev/null && echo sb_active || echo sb_inactive; " + f"{s}iptables -S SCANNER-BLOCK 2>/dev/null | grep -c DROP" + ) + rc, out, err = ssh_exec(server, cmd) + if rc != 0 or "ok" not in out: + status["error"] = (err or out or "unreachable").strip() + return status + status["reachable"] = True + lines = [l.strip() for l in out.splitlines() if l.strip()] + try: + status["country_cidrs"] = int(lines[1]) if len(lines) > 1 else 0 + except (ValueError, IndexError): + status["country_cidrs"] = 0 + if len(lines) > 2: + status["country_block"] = lines[2] == "cb_active" + if len(lines) > 3: + status["scanner_block"] = lines[3] == "sb_active" + if len(lines) > 4: + try: + status["scanner_ips"] = int(lines[4]) + except ValueError: + pass + + _fw_cache[key] = (now, status) + return status + + +def block_ip(server, ip): + s = _sudo(server) + # validate IP to avoid shell injection + if not ip or any(c not in "0123456789." for c in ip): + return {"error": "invalid IP"} + rc, out, err = ssh_exec(server, f"{s}iptables -C SCANNER-BLOCK -s {ip} -j DROP 2>/dev/null || {s}iptables -A SCANNER-BLOCK -s {ip} -j DROP") + if rc != 0: + return {"error": (err or out).strip()} + return {"status": "blocked", "ip": ip} + + +def unblock_ip(server, ip): + s = _sudo(server) + if not ip or any(c not in "0123456789." for c in ip): + return {"error": "invalid IP"} + rc, out, err = ssh_exec(server, f"{s}iptables -D SCANNER-BLOCK -s {ip} -j DROP 2>/dev/null") + if rc != 0: + return {"error": (err or out).strip()} + return {"status": "unblocked", "ip": ip} + + +def toggle_country(server, enabled): + s = _sudo(server) + rule_in = f"{s}iptables -C INPUT -m set --match-set blocked_countries src -j DROP 2>/dev/null" + rule_du = f"{s}iptables -C DOCKER-USER -m set --match-set blocked_countries src -j DROP 2>/dev/null" + add_in = f"{s}iptables -I INPUT 1 -m set --match-set blocked_countries src -j DROP" + add_du = f"{s}iptables -I DOCKER-USER 1 -m set --match-set blocked_countries src -j DROP" + del_in = f"{s}iptables -D INPUT -m set --match-set blocked_countries src -j DROP 2>/dev/null" + del_du = f"{s}iptables -D DOCKER-USER -m set --match-set blocked_countries src -j DROP 2>/dev/null" + if enabled: + cmd = f"({rule_in}) || {add_in}; ({rule_du}) || {add_du}" + else: + cmd = f"{del_in}; {del_du}" + rc, out, err = ssh_exec(server, cmd) + if rc != 0: + return {"error": (err or out).strip()} + return {"status": "enabled" if enabled else "disabled", "type": "country"} + + +def toggle_scanner(server, enabled): + s = _sudo(server) + rule_in = f"{s}iptables -C INPUT -j SCANNER-BLOCK 2>/dev/null" + rule_du = f"{s}iptables -C DOCKER-USER -j SCANNER-BLOCK 2>/dev/null" + add_in = f"{s}iptables -I INPUT 1 -j SCANNER-BLOCK" + add_du = f"{s}iptables -I DOCKER-USER 1 -j SCANNER-BLOCK" + del_in = f"{s}iptables -D INPUT -j SCANNER-BLOCK 2>/dev/null" + del_du = f"{s}iptables -D DOCKER-USER -j SCANNER-BLOCK 2>/dev/null" + if enabled: + cmd = f"({rule_in}) || {add_in}; ({rule_du}) || {add_du}" + else: + cmd = f"{del_in}; {del_du}" + rc, out, err = ssh_exec(server, cmd) + if rc != 0: + return {"error": (err or out).strip()} + return {"status": "enabled" if enabled else "disabled", "type": "scanner"} diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..8bb1749 --- /dev/null +++ b/app/main.py @@ -0,0 +1,296 @@ +import json +import os +import re +import threading +import csv +from io import StringIO +from collections import Counter +from contextlib import asynccontextmanager +from datetime import datetime +from pathlib import Path + +from fastapi import FastAPI, Query, Request +from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse +from fastapi.templating import Jinja2Templates + +from app import security +from app import firewall +from app import vps +from app import docker_ops + +LOGS_DIR = Path(os.environ.get("LOGS_DIR", "/logs")) +CONFIG_DIR = Path(os.environ.get("CONFIG_DIR", "/config")) +PROXIES_FILE = CONFIG_DIR / "proxies.json" + + +@asynccontextmanager +async def lifespan(app: FastAPI): + if os.environ.get("WEBHOOK_URL"): + t = threading.Thread(target=security.alert_loop, daemon=True) + t.start() + yield + + +app = FastAPI(title="Apache Proxy Monitor", lifespan=lifespan) +templates = Jinja2Templates(directory=Path(__file__).parent / "templates") + +API_TOKEN = os.environ.get("API_TOKEN", "") + + +@app.middleware("http") +async def token_auth(request: Request, call_next): + if API_TOKEN and request.url.path.startswith("/api/"): + if request.headers.get("X-API-Token") != API_TOKEN: + return JSONResponse({"error": "unauthorized"}, status_code=401) + return await call_next(request) + +# Combined log format, with optional "%v:%p" vhost prefix (other_vhosts_access.log) +ACCESS_RE = re.compile( + r"^(?:(?P\S+:\d+)\s+)?" + r"(?P\S+)\s+\S+\s+\S+\s+" + r"\[(?P[^\]]+)\]\s+" + r'"(?P\S+)\s+(?P\S+)(?:\s+(?P[^"]*))?"\s+' + r"(?P\d{3})\s+(?P\S+)\s+" + r'"(?P[^"]*)"\s+"(?P[^"]*)"' +) + +ERROR_RE = re.compile( + r"\[(?P[^\]]+)\]\s+\[(?P[^\]]+)\]\s+\[pid \d+[^\]]*\]\s*(?P.*)" +) + + +def load_vhosts(): + if not PROXIES_FILE.exists(): + return [] + try: + return json.loads(PROXIES_FILE.read_text()) + except Exception: + return [] + + +def vhost_from_filename(path): + name = path.name + for suffix in ("-access.log", "_access.log"): + if name.endswith(suffix): + return name[: -len(suffix)] + return name.replace(".log", "") + + +def parse_ts(s): + try: + return datetime.strptime(s, "%d/%b/%Y:%H:%M:%S %z") + except Exception: + return None + + +def read_requests(max_lines=200): + requests = [] + for path in sorted(LOGS_DIR.glob("*access*.log")): + fname_vhost = vhost_from_filename(path) + try: + lines = path.read_text(errors="replace").splitlines() + except Exception: + continue + for line in lines: + m = ACCESS_RE.match(line) + if not m: + continue + dt = parse_ts(m.group("ts")) + requests.append( + { + "vhost": m.group("vhost") or fname_vhost, + "ip": m.group("ip"), + "ts": m.group("ts"), + "time": dt.timestamp() if dt else 0, + "method": m.group("method"), + "path": m.group("path"), + "proto": m.group("proto") or "", + "status": m.group("status"), + "bytes": m.group("bytes"), + "referer": m.group("referer"), + "agent": m.group("agent"), + } + ) + requests.sort(key=lambda r: r["time"], reverse=True) + return requests[:max_lines] + + +@app.get("/", response_class=HTMLResponse) +async def index(request: Request): + return templates.TemplateResponse(request, "index.html", {"request": request}) + + +@app.get("/api/vhosts") +async def api_vhosts(): + return load_vhosts() + + +@app.get("/api/logs") +async def api_logs(lines: int = Query(200, ge=1, le=2000)): + return read_requests(lines) + + +@app.get("/api/logs/ip/{ip}") +async def api_logs_ip(ip: str): + reqs = read_requests(max_lines=100000) + return [r for r in reqs if r["ip"] == ip][:1000] + + +@app.get("/api/vps/logs/ip/{ip}") +async def api_vps_logs_ip(ip: str): + return vps.fetch_logs_ip(ip) + + +@app.get("/api/docker/info") +async def api_docker_info(): + return docker_ops.host_info() + + +@app.get("/api/docker/containers") +async def api_docker_containers(): + return docker_ops.list_containers() + + +@app.get("/api/docker/images") +async def api_docker_images(): + return docker_ops.list_images() + + +@app.post("/api/docker/containers/{cid}/{action}") +async def api_docker_action(cid: str, action: str): + return docker_ops.container_action(cid, action) + + +@app.get("/api/docker/containers/{cid}/stats") +async def api_docker_stats(cid: str): + return docker_ops.container_stats(cid) + + +@app.get("/api/errors") +async def api_errors(lines: int = Query(50, ge=1, le=500)): + path = LOGS_DIR / "error.log" + if not path.exists(): + return [] + try: + raw = path.read_text(errors="replace").splitlines() + except Exception: + return [] + return raw[-lines:][::-1] + + +@app.get("/api/stats") +async def api_stats(): + reqs = read_requests(max_lines=100000) + vhosts = load_vhosts() + status_counts = Counter(r["status"] for r in reqs) + ip_counts = Counter(r["ip"] for r in reqs) + vhost_counts = Counter(r["vhost"] for r in reqs) + return { + "total": len(reqs), + "unique_ips": len(ip_counts), + "status": dict(status_counts.most_common()), + "top_ips": [{"ip": ip, "count": c} for ip, c in ip_counts.most_common(15)], + "per_vhost": [{"vhost": v, "count": c} for v, c in vhost_counts.most_common()], + "errors_4xx": sum(c for s, c in status_counts.items() if s.startswith("4")), + "errors_5xx": sum(c for s, c in status_counts.items() if s.startswith("5")), + "vhosts": vhosts, + } + + +@app.get("/api/security") +async def api_security(hours: int = Query(168, ge=1, le=720)): + return security.security_summary(hours=hours) + + +@app.get("/api/security/ip/{ip}") +async def api_security_ip(ip: str): + return security.ip_attempts(ip) + + +@app.get("/api/geo/{ip}") +async def api_geo(ip: str): + return security.geolocate([ip]).get(ip, {}) + + +@app.post("/api/block/{ip}") +async def api_block_ip(ip: str): + results = [] + for s in firewall.load_servers(): + if s.get("host") == "local": + continue + results.append({"server": s.get("name"), "result": firewall.block_ip(s, ip)}) + results.append({"server": "ubuntu VPS (.2)", "result": vps.block_ip(ip)}) + return results + + +@app.get("/api/threats/export") +async def api_threats_export(): + sec = security.security_summary(hours=720) + buf = StringIO() + w = csv.writer(buf) + w.writerow(["ip", "country", "cc", "org", "asn", "attempts", "first_seen", "last_seen"]) + for a in sec.get("attempts", []): + w.writerow([a["ip"], a["country"], a["cc"], a["org"], a["as"], a["count"], a["first"], a["last"]]) + return PlainTextResponse(buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=threats.csv"}) + + +@app.get("/api/security/nodes") +async def api_security_nodes(): + return security.registered_nodes() + + +@app.get("/api/security/status") +async def api_security_status(): + return security.compromise_status() + + +@app.get("/api/firewall") +async def api_firewall(): + results = [firewall.firewall_status(s) for s in firewall.load_servers()] + try: + agent_fw = vps.fetch_firewall() + agent_names = {a.get("name") for a in agent_fw} + results = [r for r in results if r.get("name") not in agent_names] + agent_fw + except Exception: + pass + return results + + +@app.post("/api/firewall/{server_id}/block") +async def api_firewall_block(server_id: int, data: dict): + servers = firewall.load_servers() + if server_id < 0 or server_id >= len(servers): + raise HTTPException(404, "Server not found") + return firewall.block_ip(servers[server_id], data.get("ip", "")) + + +@app.post("/api/firewall/{server_id}/unblock") +async def api_firewall_unblock(server_id: int, data: dict): + servers = firewall.load_servers() + if server_id < 0 or server_id >= len(servers): + raise HTTPException(404, "Server not found") + return firewall.unblock_ip(servers[server_id], data.get("ip", "")) + + +@app.post("/api/firewall/{server_id}/toggle") +async def api_firewall_toggle(server_id: int, data: dict): + servers = firewall.load_servers() + if server_id < 0 or server_id >= len(servers): + raise HTTPException(404, "Server not found") + t = data.get("type") + enabled = bool(data.get("enabled", True)) + if t == "country": + return firewall.toggle_country(servers[server_id], enabled) + if t == "scanner": + return firewall.toggle_scanner(servers[server_id], enabled) + raise HTTPException(400, "Unknown toggle type") + + +@app.get("/api/vps/logs") +async def api_vps_logs(lines: int = Query(500, ge=1, le=5000)): + return vps.fetch_logs(lines) + + +@app.get("/api/vps/stats") +async def api_vps_stats(): + return vps.fetch_stats() diff --git a/app/security.py b/app/security.py new file mode 100644 index 0000000..1c0e57e --- /dev/null +++ b/app/security.py @@ -0,0 +1,388 @@ +import json +import os +import re +import threading +import time +import urllib.request +import glob +import gzip +import ipaddress +from collections import Counter, defaultdict +from datetime import datetime, timedelta +from pathlib import Path + +try: + import docker +except ImportError: + docker = None + +GEO_CACHE = Path(os.environ.get("GEO_CACHE", "/cache/geo.json")) +HEADSCALE_CONTAINER = os.environ.get("HEADSCALE_CONTAINER", "headscale") + +_client = None +_cache = {} + + +def _ttl(seconds): + def deco(fn): + def wrapper(*a, **kw): + key = fn.__name__ + now = time.time() + item = _cache.get(key) + if item and now - item[0] < seconds: + return item[1] + val = fn(*a, **kw) + _cache[key] = (now, val) + return val + return wrapper + return deco + + +def _docker_client(): + global _client + if docker is None: + return None + if _client is None: + _client = docker.DockerClient(base_url="unix:///var/run/docker.sock", timeout=30) + return _client + + +# Two timestamp formats appear in headscale logs (old + new) +TLS_ERR_RE = re.compile( + r"(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) .*TLS handshake error from (\d+\.\d+\.\d+\.\d+):\d+" +) +AUTH_ERR_RE = re.compile( + r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z) .*client_address=(\d+\.\d+\.\d+\.\d+):\d+" +) + +PRIVATE_PREFIXES = ("10.", "172.", "192.168.", "127.", "100.64.") + + +def _is_public(ip): + return not ip.startswith(PRIVATE_PREFIXES) + + +def _load_geo_cache(): + if GEO_CACHE.exists(): + try: + return json.loads(GEO_CACHE.read_text()) + except Exception: + return {} + return {} + + +def _save_geo_cache(cache): + try: + GEO_CACHE.parent.mkdir(parents=True, exist_ok=True) + GEO_CACHE.write_text(json.dumps(cache)) + except Exception: + pass + + +def geolocate(ips): + cache = _load_geo_cache() + missing = [ip for ip in ips if ip not in cache and _is_public(ip)] + for i in range(0, len(missing), 100): + batch = missing[i : i + 100] + try: + req = urllib.request.Request( + "http://ip-api.com/batch", + data=json.dumps(batch).encode(), + headers={"Content-Type": "application/json"}, + ) + data = json.loads(urllib.request.urlopen(req, timeout=20).read()) + for ip, info in zip(batch, data): + if info and info.get("status") == "success": + cache[ip] = { + "country": info.get("country"), + "cc": info.get("countryCode"), + "org": info.get("org") or info.get("isp"), + "as": info.get("as"), + } + except Exception: + pass + time.sleep(1.1) + if missing: + _save_geo_cache(cache) + return cache + + +def fetch_headscale_logs(hours=168, tail=300000): + client = _docker_client() + if client is None: + return None + try: + c = client.containers.get(HEADSCALE_CONTAINER) + since = datetime.utcnow() - timedelta(hours=hours) + return c.logs(since=since, tail=tail, timestamps=False).decode(errors="replace") + except Exception as e: + return None + + +def parse_attempts(logs): + attempts = defaultdict(lambda: {"count": 0, "first": None, "last": None}) + for line in logs.splitlines(): + m = TLS_ERR_RE.search(line) or AUTH_ERR_RE.search(line) + if not m: + continue + ts_str, ip = m.group(1), m.group(2) + a = attempts[ip] + a["count"] += 1 + a["last"] = ts_str + if a["first"] is None: + a["first"] = ts_str + return attempts + + +def _flagged_countries(): + raw = os.environ.get("FLAGGED_COUNTRIES", "China,Russia,Iran,North Korea") + return {c.strip() for c in raw.split(",") if c.strip()} + + +def send_webhook(url, text): + try: + req = urllib.request.Request( + url, + data=json.dumps({"content": text}).encode(), + headers={"Content-Type": "application/json"}, + ) + urllib.request.urlopen(req, timeout=10) + return True + except Exception as e: + print("webhook error:", e) + return False + + +def check_alerts(): + webhook = os.environ.get("WEBHOOK_URL", "") + if not webhook: + return {"status": "disabled"} + threshold = int(os.environ.get("ALERT_THRESHOLD", "50")) + state_file = Path("/cache/alerts_state.json") + + geo_cache = _load_geo_cache() + known = set(geo_cache.keys()) + state = json.loads(state_file.read_text()) if state_file.exists() else {} + + logs = fetch_headscale_logs(24, 200000) + if logs is None: + return {"status": "error", "msg": "headscale unreachable"} + attempts = parse_attempts(logs) + + geo = geolocate(list(attempts.keys())) + fired = [] + + for ip, a in attempts.items(): + if not _is_public(ip) or ip == "73.201.12.182": + continue + g = geo.get(ip, {}) + entry = state.setdefault(ip, {}) + if ip not in known and not entry.get("seen"): + entry["seen"] = True + send_webhook(webhook, f"🆕 New IP probing headscale: {ip} — {g.get('country','?')} / {g.get('org','?')} — {a['count']} attempts") + fired.append({"ip": ip, "type": "new"}) + if a["count"] >= threshold and not entry.get("thresh"): + entry["thresh"] = True + send_webhook(webhook, f"⚠️ High attempts: {ip} — {g.get('country','?')} / {g.get('org','?')} — {a['count']} attempts") + fired.append({"ip": ip, "type": "threshold"}) + + state_file.parent.mkdir(parents=True, exist_ok=True) + state_file.write_text(json.dumps(state)) + return {"status": "ok", "fired": fired} + + +def alert_loop(): + interval = int(os.environ.get("ALERT_INTERVAL", "300")) + while True: + try: + check_alerts() + except Exception as e: + print("alert loop error:", e) + time.sleep(interval) + + +def security_summary(hours=168, tail=300000): + logs = fetch_headscale_logs(hours, tail) + if logs is None: + return {"error": "headscale container not reachable"} + attempts = parse_attempts(logs) + geo = geolocate(list(attempts.keys())) + + total_attempts = sum(a["count"] for a in attempts.values()) + public = [(ip, a) for ip, a in attempts.items() if _is_public(ip)] + + rows = [] + for ip, a in public: + g = geo.get(ip, {}) + rows.append( + { + "ip": ip, + "country": g.get("country", "?"), + "cc": g.get("cc", ""), + "org": g.get("org", "?"), + "as": g.get("as", "?"), + "count": a["count"], + "first": a["first"], + "last": a["last"], + } + ) + rows.sort(key=lambda r: r["count"], reverse=True) + + flagged = _flagged_countries() + for r in rows: + r["flagged"] = r["country"] in flagged + flagged_attempts = sum(r["count"] for r in rows if r["flagged"]) + flagged_ips = sum(1 for r in rows if r["flagged"]) + + country_counts = Counter(r["country"] for r in rows) + org_counts = Counter(r["org"] for r in rows if r["org"] and r["org"] != "?") + + return { + "window_hours": hours, + "total_attempts": total_attempts, + "unique_ips": len(rows), + "countries": [{"country": c, "count": n} for c, n in country_counts.most_common()], + "top_orgs": [{"org": o, "count": n} for o, n in org_counts.most_common(15)], + "flagged": {"countries": sorted(flagged), "attempts": flagged_attempts, "ips": flagged_ips}, + "attempts": rows[:300], + } + + +@_ttl(60) +def ip_attempts(ip, hours=168): + logs = fetch_headscale_logs(hours, 300000) + if logs is None: + return {"ip": ip, "count": 0, "attempts": [], "error": "headscale unreachable"} + attempts = [] + for line in logs.splitlines(): + m = TLS_ERR_RE.search(line) or AUTH_ERR_RE.search(line) + if not m: + continue + ts, src = m.group(1), m.group(2) + if src != ip: + continue + evt = "TLS handshake failure" if "TLS" in line else ("auth error" if "client_address" in line else "connection attempt") + attempts.append({"time": ts, "event": evt}) + attempts.reverse() + return {"ip": ip, "count": len(attempts), "attempts": attempts[:500]} + + +def registered_nodes(): + client = _docker_client() + if client is None: + return [] + try: + c = client.containers.get(HEADSCALE_CONTAINER) + exit_code, output = c.exec_run(["headscale", "nodes", "list", "-o", "json"]) + if exit_code != 0: + return [] + return json.loads(output.decode(errors="replace")) + except Exception: + return [] + + +HOSTLOGS_DIR = Path(os.environ.get("HOSTLOGS_DIR", "/hostlogs")) +ACCEPTED_RE = re.compile(r"Accepted (?:password|publickey|keyboard-interactive) for (\S+) from ([0-9.]+)") +FAILED_RE = re.compile(r"Failed password for (?:invalid user )?(\S+) from ([0-9.]+)") + + +@_ttl(30) +def ssh_logins(): + accepted = Counter() + failed = Counter() + for path in sorted(glob.glob(str(HOSTLOGS_DIR / "auth.log*"))): + try: + if path.endswith(".gz"): + content = gzip.open(path, "rt", errors="replace").read() + else: + content = open(path, errors="replace").read() + except Exception: + continue + for line in content.splitlines(): + m = ACCEPTED_RE.search(line) + if m: + accepted[m.group(2)] += 1 + continue + m = FAILED_RE.search(line) + if m: + failed[m.group(2)] += 1 + return { + "accepted": [{"ip": ip, "count": c} for ip, c in accepted.most_common(20)], + "failed": [{"ip": ip, "count": c} for ip, c in failed.most_common(20)], + "total_accepted": sum(accepted.values()), + "total_failed": sum(failed.values()), + } + + +@_ttl(120) +def nextcloud_status(): + client = _docker_client() + if client is None: + return {"error": "docker not available"} + try: + c = client.containers.get("nextcloud") + except Exception: + return {"error": "nextcloud container not found"} + result = {"users": [], "failed_logins": [], "total_failed": 0} + try: + _, out = c.exec_run( + ["sh", "-c", "grep -iE 'Login failed' /var/www/html/data/nextcloud.log 2>/dev/null | tail -100"], + stdout=True, + ) + for line in out.decode(errors="replace").splitlines(): + try: + d = json.loads(line) + result["failed_logins"].append({ + "time": d.get("time"), + "ip": d.get("remoteAddr"), + "user": d.get("user"), + "message": d.get("message"), + }) + except Exception: + pass + result["total_failed"] = len(result["failed_logins"]) + except Exception: + pass + try: + _, out = c.exec_run( + ["sh", "-c", "cd /var/www/html && php occ user:list --no-warnings 2>/dev/null"], + user="www-data", stdout=True, + ) + for line in out.decode(errors="replace").splitlines(): + m = re.search(r"-\s+(\S+):", line) + if m: + result["users"].append(m.group(1)) + except Exception: + pass + return result + + +def _is_local(ip): + try: + a = ipaddress.ip_address(ip) + return a.is_private or a.is_loopback or a.is_link_local or ip == "73.201.12.182" + except ValueError: + return False + + +def compromise_status(): + ssh = ssh_logins() + nodes = registered_nodes() + nc = nextcloud_status() + warnings = [] + + unknown_ssh = [a["ip"] for a in ssh["accepted"] if not _is_local(a["ip"])] + if unknown_ssh: + warnings.append(f"SSH logins from unknown IPs: {', '.join(unknown_ssh)}") + if ssh["total_failed"] > 100: + warnings.append(f"High SSH failed-password attempts: {ssh['total_failed']}") + if nc.get("total_failed", 0) > 20: + warnings.append(f"Nextcloud failed logins: {nc['total_failed']}") + + return { + "verdict": "review" if warnings else "clean", + "warnings": warnings, + "ssh": ssh, + "headscale_nodes": nodes, + "nextcloud": nc, + } diff --git a/app/templates/index.html b/app/templates/index.html new file mode 100644 index 0000000..1dbccb9 --- /dev/null +++ b/app/templates/index.html @@ -0,0 +1,617 @@ + + + + + +DockVisor — Firewall Monitor + + + + + + + + +
+
+
+

Overview

+
Security & firewall status across all servers
+
+
+ + +
+
+
Loading…
+
+ + + + + + + diff --git a/app/vps.py b/app/vps.py new file mode 100644 index 0000000..fd55878 --- /dev/null +++ b/app/vps.py @@ -0,0 +1,75 @@ +import json +import os +import time +import urllib.request + +AGENT_URL = os.environ.get("VPS_AGENT_URL", "http://100.64.0.2:8787") +AGENT_TOKEN = os.environ.get("VPS_AGENT_TOKEN", "") + +_cache = {} + + +def _cached(key, ttl, fn): + now = time.time() + item = _cache.get(key) + if item and now - item[0] < ttl: + return item[1] + val = fn() + _cache[key] = (now, val) + return val + + +def _agent_get(path): + req = urllib.request.Request( + f"{AGENT_URL}{path}", + headers={"X-API-Token": AGENT_TOKEN}, + ) + with urllib.request.urlopen(req, timeout=20) as r: + return json.loads(r.read().decode()) + + +def fetch_logs(lines=500): + def _load(): + try: + return _agent_get(f"/api/logs?lines={lines}") + except Exception as e: + return {"error": str(e)} + return _cached(f"logs:{lines}", 30, _load) + + +def fetch_stats(): + def _load(): + try: + return _agent_get("/api/stats") + except Exception as e: + return {"error": str(e)} + return _cached("stats", 30, _load) + + +def fetch_firewall(): + def _load(): + try: + return _agent_get("/api/firewall") + except Exception as e: + return [{"name": "ubuntu VPS (.2)", "host": AGENT_URL, "reachable": False, "error": str(e)}] + return _cached("firewall", 30, _load) + + +def fetch_logs_ip(ip): + try: + return _agent_get(f"/api/logs/ip/{ip}") + except Exception as e: + return {"error": str(e)} + + +def block_ip(ip): + try: + req = urllib.request.Request( + f"{AGENT_URL}/api/firewall/0/block", + data=json.dumps({"ip": ip}).encode(), + headers={"Content-Type": "application/json", "X-API-Token": AGENT_TOKEN}, + ) + with urllib.request.urlopen(req, timeout=15) as r: + return json.loads(r.read().decode()) + except Exception as e: + return {"error": str(e)} diff --git a/docker-compose.agent.yml b/docker-compose.agent.yml new file mode 100644 index 0000000..152e90d --- /dev/null +++ b/docker-compose.agent.yml @@ -0,0 +1,21 @@ +services: + dockvisor-agent: + build: + context: . + dockerfile: Dockerfile.agent + container_name: dockvisor-agent + network_mode: host + cap_add: + - NET_ADMIN + volumes: + - /var/log/apache2:/logs:ro + - /var/log:/hostlogs:ro + - /data/dockvisor-agent:/cache + environment: + - API_TOKEN=L-04xXAHo3mN-q9N27H3Wwe1hSQKynm_rZHtjnKYJ0o + - LOGS_DIR=/logs + - HOSTLOGS_DIR=/hostlogs + - SERVERS_FILE=/cache/servers.json + - GEO_CACHE=/cache/geo.json + - HEADSCALE_CONTAINER= + restart: unless-stopped diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..7187321 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,24 @@ +services: + proxy-monitor: + build: . + container_name: proxy-monitor + network_mode: host + command: ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "9091"] + volumes: + - /data/apache-proxy-logs:/logs:ro + - /data/apache-proxy:/config:ro + - /var/run/docker.sock:/var/run/docker.sock:ro + - /data/proxy-monitor-cache:/cache + - /var/log:/hostlogs:ro + - ~/.ssh:/root/.ssh:ro + environment: + - GEO_CACHE=/cache/geo.json + - HEADSCALE_CONTAINER=headscale + - FLAGGED_COUNTRIES=China,Russia,Iran,North Korea + - VPS_AGENT_URL=http://100.64.0.2:8787 + - VPS_AGENT_TOKEN=L-04xXAHo3mN-q9N27H3Wwe1hSQKynm_rZHtjnKYJ0o + # Discord/Slack webhook URL. Leave empty to disable alerts. + - WEBHOOK_URL= + - ALERT_THRESHOLD=50 + - ALERT_INTERVAL=300 + restart: unless-stopped diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2b88449 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +fastapi +uvicorn +jinja2 +docker