297 lines
8.8 KiB
Python
297 lines
8.8 KiB
Python
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<vhost>\S+:\d+)\s+)?"
|
|
r"(?P<ip>\S+)\s+\S+\s+\S+\s+"
|
|
r"\[(?P<ts>[^\]]+)\]\s+"
|
|
r'"(?P<method>\S+)\s+(?P<path>\S+)(?:\s+(?P<proto>[^"]*))?"\s+'
|
|
r"(?P<status>\d{3})\s+(?P<bytes>\S+)\s+"
|
|
r'"(?P<referer>[^"]*)"\s+"(?P<agent>[^"]*)"'
|
|
)
|
|
|
|
ERROR_RE = re.compile(
|
|
r"\[(?P<ts>[^\]]+)\]\s+\[(?P<module>[^\]]+)\]\s+\[pid \d+[^\]]*\]\s*(?P<msg>.*)"
|
|
)
|
|
|
|
|
|
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()
|