76 lines
1.9 KiB
Python
76 lines
1.9 KiB
Python
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)}
|