716 lines
40 KiB
Python
Executable File
716 lines
40 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
AVNI Cloud — control center.
|
|
|
|
Run bare: avni -> full-screen TUI
|
|
Or CLI: avni <command>
|
|
|
|
status host overview (gauges in TUI)
|
|
services containers + status + domain
|
|
domains domains + live HTTPS + cert expiry
|
|
traffic [24h|7d|30d] bandwidth charts (vnstat) + per-container net
|
|
perdomain per-domain requests + bytes (needs access log)
|
|
bans fail2ban jails + banned IPs
|
|
attackers [N] top N SSH brute-force IPs (+country)
|
|
ipinfo <ip> full intel on one IP (country/attempts/timeline/users)
|
|
geo <ip> country of an IP
|
|
ban|unban <ip> [jail] temporary fail2ban ban/unban
|
|
blacklist [add|del <x>] PERMANENT firewall blacklist (ip|cidr|domain)
|
|
whitelist fail2ban ignoreip
|
|
ssh sshd policy + sessions(+country) + recent logins
|
|
kick <ip|pts/N> kill SSH session(s) by IP or tty
|
|
users system users: sudo/live/locked/key/lastlogin
|
|
accesslog status|on Traefik per-domain access logging
|
|
Theme: AVNI ink #0e0a09 / cream #f2ddc0
|
|
"""
|
|
import curses, subprocess, shutil, re, json, os, sys, time, glob
|
|
from datetime import datetime
|
|
|
|
IFACE_DEFAULT = None
|
|
TRAEFIK = "appwrite-traefik"
|
|
ACCESS_LOG = "/var/log/traefik/access.log"
|
|
FW = "/usr/local/bin/avni-firewall.sh"
|
|
AUTH_LOGS = ["/var/log/auth.log", "/var/log/auth.log.1"]
|
|
_GEO = {}
|
|
BLOCKS = "▁▂▃▄▅▆▇█"
|
|
|
|
# --------------------------------------------------------------------------- shell
|
|
def sh(cmd, timeout=8):
|
|
try: return subprocess.run(cmd.split(), capture_output=True, text=True, timeout=timeout).stdout.strip()
|
|
except Exception: return ""
|
|
def shl(args, timeout=8):
|
|
try: return subprocess.run(args, capture_output=True, text=True, timeout=timeout).stdout.strip()
|
|
except Exception: return ""
|
|
def sh_raw(cmd, timeout=8):
|
|
try: return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout).stdout
|
|
except Exception: return ""
|
|
def has(t): return shutil.which(t) is not None
|
|
|
|
def iface():
|
|
global IFACE_DEFAULT
|
|
if IFACE_DEFAULT: return IFACE_DEFAULT
|
|
m = re.search(r"default.* dev (\S+)", sh("ip route")); IFACE_DEFAULT = m.group(1) if m else "eth0"
|
|
return IFACE_DEFAULT
|
|
|
|
def human(n):
|
|
n=float(n)
|
|
for u in ("B","K","M","G","T"):
|
|
if n<1024: return f"{n:.0f}{u}" if u=="B" else f"{n:.1f}{u}"
|
|
n/=1024
|
|
return f"{n:.1f}P"
|
|
|
|
# --------------------------------------------------------------------------- charts (ascii)
|
|
def bar(val, mx, width, fill="█", empty="░"):
|
|
if mx<=0: return empty*width
|
|
f=int(round(width*min(val,mx)/mx)); return fill*f+empty*(width-f)
|
|
def spark(vals):
|
|
if not vals: return ""
|
|
mx=max(vals) or 1
|
|
return "".join(BLOCKS[min(len(BLOCKS)-1,int((v/mx)*(len(BLOCKS)-1)))] for v in vals)
|
|
|
|
# --------------------------------------------------------------------------- geo / intel
|
|
def geo(ip):
|
|
if ip in _GEO: return _GEO[ip]
|
|
c="?"
|
|
if has("geoiplookup"):
|
|
out=sh(f"geoiplookup {ip}")
|
|
m=re.search(r"GeoIP Country Edition:\s*[A-Z]{2},\s*(.+)", out)
|
|
if m: c=m.group(1).strip()
|
|
elif "not found" in out.lower() or "can't" in out.lower(): c="—"
|
|
_GEO[ip]=c; return c
|
|
|
|
def ts_of(line):
|
|
m=re.match(r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})", line)
|
|
if m: return m.group(1).replace("T"," ")
|
|
m=re.match(r"([A-Z][a-z]{2}\s+\d+\s\d{2}:\d{2}:\d{2})", line)
|
|
return m.group(1) if m else "?"
|
|
|
|
def ip_intel(ip):
|
|
cnt=0; first=last=None; users={}
|
|
ipre=re.compile(re.escape(ip))
|
|
ure=re.compile(r"(?:Invalid user|user)\s+([A-Za-z0-9._-]+)")
|
|
for f in AUTH_LOGS:
|
|
if not os.path.exists(f): continue
|
|
try:
|
|
with open(f, errors="ignore") as fh:
|
|
for line in fh:
|
|
if ipre.search(line) and re.search(r"Failed|Invalid|authenticating|closed by|Disconnect", line):
|
|
cnt+=1; t=ts_of(line)
|
|
if first is None or t<first: first=t
|
|
if last is None or t>last: last=t
|
|
mu=ure.search(line)
|
|
if mu: users[mu.group(1)]=users.get(mu.group(1),0)+1
|
|
except Exception: pass
|
|
banned = any(ip in f2b_status(j)["ips"] for j in f2b_jails())
|
|
bl = subprocess.run(["ipset","test","avni_blacklist",ip],capture_output=True).returncode==0
|
|
org=""
|
|
if has("whois"):
|
|
w=sh(f"whois {ip}",10)
|
|
mo=re.search(r"(?:OrgName|org-name|netname|descr):\s*(.+)", w, re.I)
|
|
if mo: org=mo.group(1).strip()
|
|
topu=sorted(users.items(),key=lambda x:-x[1])[:6]
|
|
return {"ip":ip,"country":geo(ip),"attempts":cnt,"first":first or "—","last":last or "—",
|
|
"users":topu,"banned":banned,"blacklisted":bl,"org":org}
|
|
|
|
# --------------------------------------------------------------------------- docker / services
|
|
def overview():
|
|
mem=next((l for l in sh_raw("free -m").splitlines() if l.lower().startswith("mem")),"").split()
|
|
mtot,mused,mavail=(int(mem[1]),int(mem[2]),int(mem[-1])) if len(mem)>=7 else (0,0,0)
|
|
d=sh("df -h /").splitlines(); dl=d[1].split() if len(d)>1 else [""]*6
|
|
du=sh("df / ").splitlines(); dpct=du[1].split()[4] if len(du)>1 else "0%"
|
|
load=open("/proc/loadavg").read().split()[:3]
|
|
return {"host":sh("hostname"),"ip":(sh("curl -s -m4 https://api.ipify.org") or ""),
|
|
"os":sh_raw(". /etc/os-release; echo $PRETTY_NAME").strip(),"kernel":sh("uname -r"),
|
|
"uptime":sh("uptime -p"),"load":load,"ncpu":os.cpu_count() or 1,
|
|
"mem_total":mtot,"mem_used":mused,"mem_avail":mavail,
|
|
"disk_size":dl[1],"disk_used":dl[2],"disk_avail":dl[3],"disk_pct":dl[4]}
|
|
|
|
def traefik_domains():
|
|
out={}; ids=sh("docker ps -q").split()
|
|
if not ids: return out
|
|
try: data=json.loads(shl(["docker","inspect"]+ids,15) or "[]")
|
|
except Exception: return out
|
|
for c in data:
|
|
name=c.get("Name","").lstrip("/"); labels=(c.get("Config",{}) or {}).get("Labels",{}) or {}
|
|
for k,v in labels.items():
|
|
if k.endswith(".rule") and "Host(" in v:
|
|
for d in re.findall(r"Host\(`([^`]+)`\)",v): out.setdefault(d,name)
|
|
return out
|
|
|
|
def services():
|
|
rows=[]
|
|
out=shl(["docker","ps","-a","--format","{{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}"],12)
|
|
for line in out.splitlines():
|
|
p=line.split("\t")
|
|
if len(p)<3: continue
|
|
rows.append({"name":p[0],"image":p[1],"status":p[2],"ports":p[3] if len(p)>3 else ""})
|
|
c2d={}
|
|
for d,c in traefik_domains().items(): c2d.setdefault(c,[]).append(d)
|
|
for r in rows: r["domain"]=",".join(c2d.get(r["name"],[]))
|
|
return rows
|
|
|
|
def docker_net():
|
|
rows=[]
|
|
for line in shl(["docker","stats","--no-stream","--format","{{.Name}}\t{{.NetIO}}\t{{.CPUPerc}}"],20).splitlines():
|
|
p=line.split("\t")
|
|
if len(p)>=2: rows.append({"name":p[0],"net":p[1],"cpu":p[2] if len(p)>2 else ""})
|
|
return rows
|
|
|
|
def cert_expiry(domain):
|
|
out=sh_raw(f"echo | timeout 6 openssl s_client -servername {domain} -connect {domain}:443 2>/dev/null | openssl x509 -noout -enddate 2>/dev/null")
|
|
m=re.search(r"notAfter=(.+)",out); return m.group(1).strip() if m else "?"
|
|
|
|
# --------------------------------------------------------------------------- fail2ban / blacklist
|
|
def f2b_jails():
|
|
m=re.search(r"Jail list:\s*(.*)",sh("fail2ban-client status"))
|
|
return [j.strip() for j in m.group(1).split(",")] if m and m.group(1).strip() else []
|
|
def f2b_status(jail):
|
|
out=sh(f"fail2ban-client status {jail}")
|
|
g=lambda p,d="0":(re.search(p,out).group(1) if re.search(p,out) else d)
|
|
ips=re.search(r"Banned IP list:\s*(.*)",out)
|
|
return {"failed":g(r"Currently failed:\s*(\d+)"),"banned":g(r"Currently banned:\s*(\d+)"),
|
|
"total":g(r"Total banned:\s*(\d+)"),"ips":ips.group(1).split() if ips and ips.group(1).strip() else []}
|
|
def f2b_banned_all():
|
|
return [{"jail":j,"ip":ip} for j in f2b_jails() for ip in f2b_status(j)["ips"]]
|
|
def f2b_ignoreip(jail="sshd"): return sh(f"fail2ban-client get {jail} ignoreip")
|
|
def ban(ip,jail="sshd"): return sh(f"fail2ban-client set {jail} banip {ip}")
|
|
def unban(ip,jail="sshd"): return sh(f"fail2ban-client set {jail} unbanip {ip}")
|
|
|
|
def blacklist_list(): return [x for x in sh(f"bash {FW} list").splitlines() if x.strip()]
|
|
def blacklist_add(x): return sh(f"bash {FW} add {x}",15)
|
|
def blacklist_del(x): return sh(f"bash {FW} del {x}",15)
|
|
def blacklist_count():
|
|
try: return int(sh(f"bash {FW} count") or "0")
|
|
except Exception: return 0
|
|
|
|
def attackers(n=15):
|
|
counts={}
|
|
pat=re.compile(r"(Failed password|Invalid user|authenticating user|Connection closed by|Disconnected from invalid)")
|
|
ipre=re.compile(r"(\d{1,3}\.){3}\d{1,3}")
|
|
for f in AUTH_LOGS:
|
|
if not os.path.exists(f): continue
|
|
try:
|
|
with open(f,errors="ignore") as fh:
|
|
for line in fh:
|
|
if pat.search(line):
|
|
m=ipre.search(line)
|
|
if m: counts[m.group(0)]=counts.get(m.group(0),0)+1
|
|
except Exception: pass
|
|
return [{"ip":ip,"count":c} for ip,c in sorted(counts.items(),key=lambda x:-x[1])[:n]]
|
|
|
|
def whois_ip(ip):
|
|
if not has("whois"): return "install: apt-get install -y whois"
|
|
keep=[l for l in sh(f"whois {ip}",10).splitlines() if re.match(r"^\s*(country|orgname|org-name|netname|descr|inetnum):",l,re.I)]
|
|
return "\n".join(keep[:8]) or "(no whois data)"
|
|
|
|
# --------------------------------------------------------------------------- IP registry (CrowdSec-style decisions store)
|
|
REG_PATH="/var/lib/avni/ips.json"
|
|
def registry_load():
|
|
try: return json.load(open(REG_PATH))
|
|
except Exception: return {}
|
|
def registry_save(d):
|
|
os.makedirs("/var/lib/avni",exist_ok=True)
|
|
try: json.dump(d,open(REG_PATH,"w"))
|
|
except Exception: pass
|
|
def registry_ips(reg): return [(i,e) for i,e in reg.items() if not i.startswith("_")]
|
|
|
|
def _scan_authlogs(offsets):
|
|
"""Read only NEW lines since last byte-offset (keyed by inode) -> fast."""
|
|
att={}; fs={}; ls={}; users={}; new=dict(offsets)
|
|
pat=re.compile(r"(Failed password|Invalid user|authenticating|closed by|Disconnect)")
|
|
ipre=re.compile(r"(\d{1,3}\.){3}\d{1,3}"); ure=re.compile(r"(?:Invalid user|user)\s+([A-Za-z0-9._-]+)")
|
|
for f in AUTH_LOGS:
|
|
if not os.path.exists(f): continue
|
|
try:
|
|
st=os.stat(f); ino=str(st.st_ino); start=offsets.get(ino,0)
|
|
if start>st.st_size: start=0 # rotated/truncated
|
|
with open(f,errors="ignore") as fh:
|
|
fh.seek(start)
|
|
for line in fh:
|
|
if not pat.search(line): continue
|
|
m=ipre.search(line)
|
|
if not m: continue
|
|
ip=m.group(0); t=ts_of(line); att[ip]=att.get(ip,0)+1
|
|
if ip not in fs or t<fs[ip]: fs[ip]=t
|
|
if ip not in ls or t>ls[ip]: ls[ip]=t
|
|
mu=ure.search(line)
|
|
if mu: users[ip]=mu.group(1)
|
|
new[ino]=fh.tell()
|
|
except Exception: pass
|
|
return att,fs,ls,users,new
|
|
|
|
def registry_update(full=False):
|
|
"""Incremental: only reads log bytes appended since last run. The persistent
|
|
store keeps every IP forever (even after logs rotate)."""
|
|
reg=registry_load(); now=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
meta=reg.get("_meta",{}); offsets={} if full else meta.get("offsets",{})
|
|
if "_meta" not in reg and registry_ips(reg) and not full:
|
|
# pre-incremental store already has counts: mark EOF, go incremental forward
|
|
for f in AUTH_LOGS:
|
|
if os.path.exists(f): offsets[str(os.stat(f).st_ino)]=os.stat(f).st_size
|
|
att,fs,ls,users=({},{},{},{})
|
|
else:
|
|
att,fs,ls,users,offsets=_scan_authlogs(offsets)
|
|
ipre=re.compile(r"(\d{1,3}\.){3}\d{1,3}")
|
|
logins={}
|
|
for line in sh("last -i -n 300").splitlines():
|
|
p=line.split()
|
|
if len(p)>=3 and ipre.fullmatch(p[2] or ""): logins[p[2]]=logins.get(p[2],0)+1
|
|
banned=set()
|
|
for j in f2b_jails(): banned|=set(f2b_status(j)["ips"])
|
|
bl=set(blacklist_list())
|
|
for ip in set(att)|set(logins)|{i for i,_ in registry_ips(reg)}|banned|bl:
|
|
e=reg.get(ip,{})
|
|
e["country"]=e.get("country") or geo(ip)
|
|
e["attempts"]=e.get("attempts",0)+att.get(ip,0)
|
|
if ip in fs: e["first"]=min(e.get("first",fs[ip]),fs[ip])
|
|
if ip in ls: e["last"]=max(e.get("last",""),ls[ip])
|
|
e["logins"]=max(e.get("logins",0),logins.get(ip,0))
|
|
e["banned"]=ip in banned; e["blacklisted"]=ip in bl
|
|
if ip in users: e["last_user"]=users[ip]
|
|
e.setdefault("first",now); e.setdefault("last",now)
|
|
reg[ip]=e
|
|
reg["_meta"]={"offsets":offsets,"updated":now}
|
|
registry_save(reg); return reg
|
|
|
|
def _recent(ts, hours=24):
|
|
try: return (datetime.now()-datetime.strptime(ts,"%Y-%m-%d %H:%M:%S")).total_seconds() < hours*3600
|
|
except Exception: return False
|
|
|
|
def registry_filter(reg, which):
|
|
v=list(reg.items())
|
|
if which=="Attackers": v=[(i,e) for i,e in v if e.get("attempts",0)>0 and not e.get("logins")]
|
|
elif which=="Banned": v=[(i,e) for i,e in v if e.get("banned") or e.get("blacklisted")]
|
|
elif which=="Unbanned": v=[(i,e) for i,e in v if (e.get("attempts",0)>0) and not e.get("banned") and not e.get("blacklisted")]
|
|
elif which=="New": v=[(i,e) for i,e in v if _recent(e.get("first",""))]
|
|
elif which=="Logins": v=[(i,e) for i,e in v if e.get("logins",0)>0]
|
|
key={"Attackers":"attempts","Logins":"logins"}.get(which)
|
|
v.sort(key=lambda x:-(x[1].get(key,0)) if key else x[1].get("last",""), reverse=bool(not key))
|
|
if key: v.sort(key=lambda x:-x[1].get(key,0))
|
|
return v
|
|
|
|
# --------------------------------------------------------------------------- ssh / users
|
|
def ssh_sessions():
|
|
s=[]
|
|
for line in sh("who").splitlines():
|
|
p=line.split()
|
|
if len(p)>=5:
|
|
frm=p[-1].strip("()")
|
|
s.append({"user":p[0],"tty":p[1],"from":frm,"country":geo(frm) if re.match(r"\d+\.\d+\.\d+\.\d+",frm) else "local","since":" ".join(p[2:4])})
|
|
return s
|
|
def ssh_info():
|
|
t=sh("sshd -T",6)
|
|
g=lambda k:next((l.split(None,1)[1] for l in t.splitlines() if l.lower().startswith(k)),"?")
|
|
recent=[l for l in sh("last -a -n 14").splitlines() if l.strip() and not l.startswith("wtmp")]
|
|
return {"PermitRootLogin":g("permitrootlogin"),"PasswordAuthentication":g("passwordauthentication"),
|
|
"PubkeyAuthentication":g("pubkeyauthentication"),"Port":g("port"),
|
|
"sessions":ssh_sessions(),"recent":recent[:12],"attacker_ips":len(attackers(99999))}
|
|
|
|
def my_tty():
|
|
try: return os.ttyname(sys.stdin.fileno()).replace("/dev/","")
|
|
except Exception: return ""
|
|
def my_ip(): return os.environ.get("SSH_CLIENT","").split()[0] if os.environ.get("SSH_CLIENT") else ""
|
|
|
|
def kick_tty(tty):
|
|
if tty==my_tty(): return "refused: that is YOUR session"
|
|
subprocess.run(["pkill","-KILL","-t",tty]); return f"kicked session {tty}"
|
|
def kick_ip(ip):
|
|
if ip==my_ip(): return "refused: that is YOUR IP"
|
|
killed=[]
|
|
for s in ssh_sessions():
|
|
if s["from"]==ip and s["tty"]!=my_tty():
|
|
subprocess.run(["pkill","-KILL","-t",s["tty"]]); killed.append(s["tty"])
|
|
return f"kicked {ip}: "+(", ".join(killed) if killed else "no live session")
|
|
|
|
def users_list():
|
|
sudoers=set()
|
|
for grp in ("sudo","admin","wheel"):
|
|
m=sh(f"getent group {grp}")
|
|
if m and ":" in m: sudoers|=set(filter(None,m.split(":")[-1].split(",")))
|
|
live={}
|
|
for line in sh("who").splitlines():
|
|
u=line.split()[0] if line.split() else ""
|
|
live[u]=live.get(u,0)+1
|
|
rows=[]
|
|
for line in sh("getent passwd").splitlines():
|
|
p=line.split(":")
|
|
if len(p)<7: continue
|
|
name,uid,home,shell=p[0],int(p[2]),p[5],p[6]
|
|
if not(uid==0 or uid>=1000) or name=="nobody" or shell.endswith(("nologin","false")): continue
|
|
ll=sh(f"lastlog -u {name}").splitlines()
|
|
last=ll[1].split(None,3)[-1] if len(ll)>1 and "Never" not in ll[1] else "Never"
|
|
locked="L" in (sh(f"passwd -S {name}").split()[1:2] or [""])[0]
|
|
rows.append({"user":name,"uid":uid,"sudo":name in sudoers or uid==0,"shell":shell,
|
|
"lastlogin":last,"locked":locked,"key":os.path.exists(f"{home}/.ssh/authorized_keys"),
|
|
"live":live.get(name,0),"groups":sh(f"id -nG {name}")})
|
|
return rows
|
|
|
|
# --------------------------------------------------------------------------- traffic series
|
|
def vnstat_series(mode):
|
|
flag={"24h":"h","7d":"d","30d":"d"}.get(mode,"d")
|
|
try: j=json.loads(sh_raw(f"vnstat --json {flag} -i {iface()}",8) or "{}")
|
|
except Exception: j={}
|
|
out=[]
|
|
try:
|
|
node=j["interfaces"][0]["traffic"]
|
|
arr=node.get("hour") or node.get("hours") or node.get("day") or node.get("days") or []
|
|
for e in arr[-(24 if mode=="24h" else (7 if mode=="7d" else 30)):]:
|
|
d=e.get("date",{}); lab=(f"{d.get('hour',e.get('time',{}).get('hour','')):>2}h" if mode=="24h"
|
|
else f"{d.get('month','')}/{d.get('day','')}")
|
|
out.append({"label":lab,"rx":e.get("rx",0),"tx":e.get("tx",0)})
|
|
except Exception: pass
|
|
return out
|
|
def vnstat_oneline():
|
|
p=sh(f"vnstat --oneline -i {iface()}").split(";")
|
|
return {"today":p[5],"rx":p[3],"tx":p[4],"rate":p[6],"month":p[10],"all":p[13]} if len(p)>=14 else {}
|
|
|
|
# --------------------------------------------------------------------------- access log
|
|
def accesslog_status():
|
|
on="accesslog" in sh(f"docker inspect {TRAEFIK} --format "+"{{json .Args}}").lower()
|
|
return {"enabled":on,"logfile":os.path.exists(ACCESS_LOG)}
|
|
def perdomain():
|
|
if not os.path.exists(ACCESS_LOG): return None
|
|
reqs,byts={},{}
|
|
try:
|
|
with open(ACCESS_LOG,errors="ignore") as fh:
|
|
for line in fh:
|
|
try: jj=json.loads(line)
|
|
except Exception: continue
|
|
h=jj.get("RequestHost","?"); reqs[h]=reqs.get(h,0)+1; byts[h]=byts.get(h,0)+int(jj.get("DownstreamContentSize",0) or 0)
|
|
except Exception: return None
|
|
return sorted([{"domain":d,"reqs":reqs[d],"bytes":byts.get(d,0)} for d in reqs],key=lambda x:-x["reqs"])
|
|
|
|
# =========================================================================== CLI
|
|
def cli(argv):
|
|
cmd=argv[0] if argv else "status"; a=argv[1:]
|
|
if cmd in("status","overview"):
|
|
o=overview(); svc=services(); up=sum(1 for s in svc if s["status"].startswith("Up"))
|
|
print(f"AVNI Cloud — {o['host']} ({o['ip']})")
|
|
print(f" OS {o['os']} · kernel {o['kernel']}")
|
|
print(f" Uptime {o['uptime']} · Load {' '.join(o['load'])} (cpus {o['ncpu']})")
|
|
print(f" RAM {o['mem_used']}/{o['mem_total']}MB · Disk {o['disk_used']}/{o['disk_size']} ({o['disk_pct']})")
|
|
print(f" Services {up}/{len(svc)} up · Domains {len(traefik_domains())} · Banned {f2b_status('sshd')['banned']} · Blacklist {blacklist_count()}")
|
|
elif cmd=="services":
|
|
for s in services():
|
|
print(f" {'●' if s['status'].startswith('Up') else '○'} {s['name']:<34}{s['status']:<26}{s['domain']}")
|
|
elif cmd=="domains":
|
|
for d,c in sorted(traefik_domains().items()):
|
|
print(f" {d:<32} HTTP {sh(f'curl -s -m8 -o /dev/null -w %{{http_code}} https://{d}/'):<4} cert:{cert_expiry(d):<22} -> {c}")
|
|
elif cmd=="traffic":
|
|
win=a[0] if a else "7d"; ol=vnstat_oneline()
|
|
print(f" today {ol.get('today','?')} (rx {ol.get('rx','')}/tx {ol.get('tx','')}) rate {ol.get('rate','')} · month {ol.get('month','')}")
|
|
for e in vnstat_series(win):
|
|
tot=e["rx"]+e["tx"]; print(f" {e['label']:>6} {human(tot):>8} {bar(tot,max((x['rx']+x['tx']) for x in vnstat_series(win)),30)}")
|
|
print(" per-container:"); [print(f" {r['name']:<32}{r['net']}") for r in docker_net()]
|
|
elif cmd=="perdomain":
|
|
rows=perdomain()
|
|
if rows is None: print(" needs access log — run: avni accesslog on")
|
|
else:
|
|
for r in rows: print(f" {r['domain']:<32}{r['reqs']:>9} req {human(r['bytes']):>9}")
|
|
elif cmd=="bans":
|
|
for j in f2b_jails():
|
|
s=f2b_status(j); print(f" [{j}] banned={s['banned']} total={s['total']}")
|
|
for ip in s["ips"]: print(f" {ip:<18}{geo(ip)}")
|
|
elif cmd=="attackers":
|
|
n=int(a[0]) if a and a[0].isdigit() else 15; rows=attackers(n); mx=rows[0]["count"] if rows else 1
|
|
for r in rows: print(f" {r['count']:>7} {r['ip']:<17}{geo(r['ip']):<16}{bar(r['count'],mx,24)}")
|
|
elif cmd=="ipinfo":
|
|
if not a: sys.exit("usage: avni ipinfo <ip>")
|
|
d=ip_intel(a[0])
|
|
print(f" IP {d['ip']} ({d['country']})")
|
|
print(f" attempts {d['attempts']} · first {d['first']} · last {d['last']}")
|
|
print(f" banned={d['banned']} blacklisted={d['blacklisted']} org={d['org']}")
|
|
print(f" usernames tried: "+", ".join(f"{u}({c})" for u,c in d['users']))
|
|
elif cmd=="ips":
|
|
which=a[0].capitalize() if a else "All"
|
|
reg=registry_update(); rows=registry_filter(reg,which) if which!="All" else sorted(reg.items(),key=lambda x:-x[1].get("attempts",0))
|
|
print(f" {which}: {len(rows)} IPs (total tracked: {len(reg)})")
|
|
print(f" {'IP':<16}{'COUNTRY':<14}{'ATT':>7}{'LOGIN':>6} STATUS LAST")
|
|
for ip,e in rows[:60]:
|
|
st="BLACKLIST" if e.get("blacklisted") else ("BANNED" if e.get("banned") else ("login" if e.get("logins") else "seen"))
|
|
print(f" {ip:<16}{(e.get('country','?') or '?')[:13]:<14}{e.get('attempts',0):>7}{e.get('logins',0):>6} {st:<10} {e.get('last','')}")
|
|
elif cmd=="geo": print(geo(a[0]) if a else "usage: avni geo <ip>")
|
|
elif cmd=="ban": print(ban(a[0],a[1] if len(a)>1 else "sshd"),"banned",a[0]) if a else sys.exit("usage")
|
|
elif cmd=="unban": print(unban(a[0],a[1] if len(a)>1 else "sshd"),"unbanned",a[0]) if a else sys.exit("usage")
|
|
elif cmd=="blacklist":
|
|
if not a: [print(" "+x+" "+geo(x.split('/')[0])) for x in blacklist_list()] or (print(" (empty)") if not blacklist_list() else None)
|
|
elif a[0]=="add" and len(a)>1: print(blacklist_add(a[1]))
|
|
elif a[0]=="del" and len(a)>1: print(blacklist_del(a[1]))
|
|
else: print("usage: avni blacklist [add|del <ip|cidr|domain>]")
|
|
elif cmd=="whitelist": [print(f" [{j}] {f2b_ignoreip(j)}") for j in f2b_jails()]
|
|
elif cmd=="whois": print(whois_ip(a[0]) if a else "usage: avni whois <ip>")
|
|
elif cmd=="ssh":
|
|
s=ssh_info()
|
|
print(f" PermitRootLogin={s['PermitRootLogin']} PasswordAuth={s['PasswordAuthentication']} Pubkey={s['PubkeyAuthentication']} Port={s['Port']}")
|
|
print(f" attacker IPs in auth.log: {s['attacker_ips']}"); print(" live sessions:")
|
|
for x in s["sessions"]: print(f" {x['user']}@{x['tty']:<8} {x['from']:<16}({x['country']}) since {x['since']}")
|
|
print(" recent:"); [print(" "+r) for r in s["recent"]]
|
|
elif cmd=="kick": print(kick_ip(a[0]) if a and re.match(r"\d+\.\d+\.\d+\.\d+",a[0]) else (kick_tty(a[0]) if a else "usage: avni kick <ip|pts/N>"))
|
|
elif cmd=="users":
|
|
print(f" {'USER':<12}{'UID':>5} SUDO LIVE LOCK KEY LASTLOGIN")
|
|
for u in users_list():
|
|
print(f" {u['user']:<12}{u['uid']:>5} {'yes ' if u['sudo'] else '- '} {u['live']:>3} {'L' if u['locked'] else '-'} {'y' if u['key'] else 'n'} {u['lastlogin']}")
|
|
elif cmd=="accesslog": accesslog_cmd(a[0] if a else "status")
|
|
else: print(__doc__)
|
|
|
|
def accesslog_cmd(sub):
|
|
st=accesslog_status()
|
|
if sub=="status": print(f" enabled={st['enabled']} logfile={st['logfile']} path={ACCESS_LOG}"); return
|
|
print(" Enable per-domain stats (recreates the Appwrite Traefik ingress):")
|
|
print(" 1) <your-traefik-stack>/docker-compose.yml traefik command, add:")
|
|
print(" - --accesslog=true\n - --accesslog.filepath=/var/log/traefik/access.log\n - --accesslog.format=json")
|
|
print(" volumes: - /var/log/traefik:/var/log/traefik")
|
|
print(" 2) cd <your-traefik-stack> && docker compose up -d traefik 3) avni perdomain")
|
|
|
|
# =========================================================================== TUI
|
|
SECTIONS=["Overview","Services","Domains","Traffic","Security","SSH","Users"]
|
|
def init_colors():
|
|
curses.start_color(); curses.use_default_colors()
|
|
C=lambda n,f:curses.init_pair(n,f if curses.COLORS>=256 else f%8,-1)
|
|
cream=223 if curses.COLORS>=256 else 3
|
|
curses.init_pair(1,cream,-1); curses.init_pair(2,234 if curses.COLORS>=256 else 0,cream)
|
|
curses.init_pair(3,71 if curses.COLORS>=256 else 2,-1); curses.init_pair(4,203 if curses.COLORS>=256 else 1,-1)
|
|
curses.init_pair(5,245 if curses.COLORS>=256 else 7,-1); curses.init_pair(6,curses.COLOR_WHITE,-1)
|
|
curses.init_pair(7,214 if curses.COLORS>=256 else 3,-1); curses.init_pair(8,110 if curses.COLORS>=256 else 6,-1)
|
|
|
|
class TUI:
|
|
def __init__(self,scr):
|
|
self.scr=scr; self.sec=0; self.sel=0; self.tw=0
|
|
self.msg="Tab/←→ sections · ↑↓ select · Enter detail · keys in footer · q quit"; self.cache={}
|
|
def data(self,k,fn,ttl=4):
|
|
now=time.time()
|
|
if k not in self.cache or now-self.cache[k][0]>ttl:
|
|
try: self.cache[k]=(now,fn())
|
|
except Exception: self.cache[k]=(now,[])
|
|
return self.cache[k][1]
|
|
def rows(self):
|
|
s=SECTIONS[self.sec]
|
|
if s=="Services": return [(svc_line(r),r) for r in self.data("svc",services)]
|
|
if s=="Domains": return [(d,{"domain":d,"cont":c}) for d,c in sorted(self.data("dom",traefik_domains,8).items())]
|
|
if s=="Security": return self.sec_rows()
|
|
if s=="SSH": return [(None,x) for x in self.data("ssh",ssh_info,5)["sessions"]]
|
|
if s=="Users": return [(None,u) for u in self.data("usr",users_list,10)]
|
|
return []
|
|
def sec_rows(self):
|
|
atk=self.data("atk",lambda:attackers(14),12); mx=atk[0]["count"] if atk else 1
|
|
rows=[]
|
|
for r in atk:
|
|
rows.append((None,{"kind":"atk","ip":r["ip"],"count":r["count"],"mx":mx}))
|
|
return rows
|
|
def clampsel(self,n): self.sel=max(0,min(self.sel,n-1)) if n else 0
|
|
|
|
def draw(self):
|
|
scr=self.scr; scr.erase(); h,w=scr.getmaxyx()
|
|
scr.attron(curses.color_pair(2)); scr.addstr(0,0," "*(w-1))
|
|
scr.addstr(0,1,"AVNI CLOUD · control center"[:w-2]); scr.attroff(curses.color_pair(2))
|
|
clk=datetime.now().strftime("%H:%M:%S")
|
|
if w>12: scr.addstr(0,w-len(clk)-1,clk,curses.color_pair(1))
|
|
x=0
|
|
for i,s in enumerate(SECTIONS):
|
|
lab=f" {i+1}.{s} "
|
|
if x+len(lab)>=w: break
|
|
scr.addstr(1,x,lab,curses.color_pair(2) if i==self.sec else curses.color_pair(5)); x+=len(lab)
|
|
scr.hline(2,0,curses.ACS_HLINE,w-1)
|
|
try: self.body(scr,3,h-2,w)
|
|
except Exception as e: scr.addstr(4,2,f"(render: {e})"[:w-3],curses.color_pair(4))
|
|
scr.attron(curses.color_pair(2)); scr.addstr(h-1,0," "*(w-1))
|
|
scr.addstr(h-1,1,self.msg[:max(0,w-22)]); scr.attroff(curses.color_pair(2))
|
|
foot=self.footer()
|
|
if w>len(foot)+2: scr.addstr(h-1,w-len(foot)-1,foot,curses.color_pair(2))
|
|
scr.refresh()
|
|
|
|
def gauge(self,scr,y,x,label,used,total,w,unit=""):
|
|
pct=(used/total) if total else 0
|
|
col=3 if pct<0.7 else (7 if pct<0.9 else 4)
|
|
scr.addstr(y,x,f"{label:<8}",curses.color_pair(1))
|
|
bw=max(8,w-30); b=bar(used,total,bw)
|
|
fill=int(round(bw*min(pct,1)))
|
|
scr.addstr(y,x+9,b[:fill],curses.color_pair(col)); scr.addstr(y,x+9+fill,b[fill:],curses.color_pair(5))
|
|
scr.addstr(y,x+10+bw,f"{pct*100:4.0f}% {used}/{total}{unit}"[:24],curses.color_pair(6))
|
|
|
|
def body(self,scr,top,bot,w):
|
|
s=SECTIONS[self.sec]
|
|
if s=="Overview": return self.b_overview(scr,top,bot,w)
|
|
if s=="Traffic": return self.b_traffic(scr,top,bot,w)
|
|
if s=="SSH": return self.b_ssh(scr,top,bot,w)
|
|
if s=="Users": return self.b_users(scr,top,bot,w)
|
|
if s=="Security": return self.b_security(scr,top,bot,w)
|
|
rows=self.rows(); self.clampsel(len(rows))
|
|
if not rows: scr.addstr(top+1,2,"(nothing)",curses.color_pair(5)); return
|
|
view=bot-top-1; start=max(0,self.sel-view+1)
|
|
for i,(line,obj) in enumerate(rows[start:start+view]):
|
|
idx=start+i; y=top+1+i; txt=line if line else str(obj)
|
|
if SECTIONS[self.sec]=="Domains": txt=f"{obj['domain']:<34} → {obj['cont']}"
|
|
scr.addstr(y,2,(txt[:w-4]).ljust(w-4) if idx==self.sel else txt[:w-4],
|
|
curses.color_pair(2) if idx==self.sel else color_for(txt))
|
|
|
|
def b_overview(self,scr,top,bot,w):
|
|
o=self.data("ov",overview,5); svc=self.data("svc",services); up=sum(1 for x in svc if x["status"].startswith("Up"))
|
|
ol=self.data("vol",vnstat_oneline,6)
|
|
info=[("Host",f"{o['host']} {o['ip']}"),("OS",f"{o['os']} · {o['kernel']}"),("Uptime",o["uptime"])]
|
|
for i,(k,v) in enumerate(info):
|
|
scr.addstr(top+1+i,2,f"{k:<8}",curses.color_pair(1)); scr.addstr(top+1+i,11,str(v)[:w-13],curses.color_pair(6))
|
|
y=top+5
|
|
la=float(o["load"][0]); self.gauge(scr,y,2,"Load",round(la,2),o["ncpu"],w,"")
|
|
self.gauge(scr,y+1,2,"Memory",o["mem_used"],o["mem_total"],w,"MB")
|
|
dpct=int(re.sub(r"\D","",o["disk_pct"]) or 0); self.gauge(scr,y+2,2,"Disk",dpct,100,w,"%")
|
|
y+=4
|
|
scr.addstr(y,2,f"Services {up}/{len(svc)} up Domains {len(self.data('dom',traefik_domains,8))} "
|
|
f"Banned {f2b_status('sshd')['banned']} Blacklist {self.data('blc',blacklist_count,10)}",curses.color_pair(1))
|
|
scr.addstr(y+2,2,f"Net today {ol.get('today','?')} (rx {ol.get('rx','')} / tx {ol.get('tx','')}) rate {ol.get('rate','')} · month {ol.get('month','')}",curses.color_pair(6))
|
|
att=self.data("atk",lambda:attackers(14),12)
|
|
scr.addstr(y+4,2,"Attack pressure (top IPs): "+spark([a["count"] for a in att]),curses.color_pair(4))
|
|
|
|
def b_traffic(self,scr,top,bot,w):
|
|
win=["24h","7d","30d"][self.tw%3]
|
|
scr.addstr(top+1,2,f"Window < {win} > (↑↓ switches 24h / 7d / 30d)",curses.color_pair(1))
|
|
ser=self.data("vn"+win,lambda:vnstat_series(win),20)
|
|
ol=self.data("vol",vnstat_oneline,6)
|
|
scr.addstr(top+2,2,f"today {ol.get('today','?')} rate {ol.get('rate','')} month {ol.get('month','')}",curses.color_pair(6))
|
|
y=top+4; mx=max((e["rx"]+e["tx"]) for e in ser) if ser else 1
|
|
for e in ser[-(bot-top-6):]:
|
|
tot=e["rx"]+e["tx"]
|
|
scr.addstr(y,2,f"{e['label']:>6} {human(tot):>8} ",curses.color_pair(6))
|
|
bw=max(6,w-40); scr.addstr(y,18,bar(tot,mx,bw),curses.color_pair(8)); y+=1
|
|
if y>=bot-1: break
|
|
if not ser: scr.addstr(y,2,"(vnstat is still collecting — history fills in over the coming days)",curses.color_pair(5))
|
|
|
|
def b_ssh(self,scr,top,bot,w):
|
|
s=self.data("ssh",ssh_info,5); warn=s["PasswordAuthentication"]=="yes"
|
|
scr.addstr(top+1,2,f"PermitRootLogin {s['PermitRootLogin']} PasswordAuth {s['PasswordAuthentication']} Pubkey {s['PubkeyAuthentication']} Port {s['Port']}",
|
|
curses.color_pair(4 if warn else 3))
|
|
scr.addstr(top+2,2,f"distinct attacker IPs in auth.log: {s['attacker_ips']}",curses.color_pair(1))
|
|
scr.addstr(top+4,2,"LIVE SESSIONS (↑↓ select · k kick session · K kick+blacklist IP):",curses.color_pair(1))
|
|
sess=s["sessions"]; self.clampsel(len(sess)); y=top+5
|
|
for i,x in enumerate(sess):
|
|
mark="»" if i==self.sel else " "; me=" (you)" if x["tty"]==my_tty() else ""
|
|
line=f"{mark} {x['user']}@{x['tty']:<9}{x['from']:<16} {x['country']:<14} since {x['since']}{me}"
|
|
scr.addstr(y,2,line[:w-4],curses.color_pair(2) if i==self.sel else curses.color_pair(6)); y+=1
|
|
y+=1; scr.addstr(y,2,"recent logins:",curses.color_pair(1)); y+=1
|
|
for r in s["recent"]:
|
|
if y>=bot-1: break
|
|
scr.addstr(y,4,r[:w-6],curses.color_pair(5)); y+=1
|
|
|
|
def b_users(self,scr,top,bot,w):
|
|
us=self.data("usr",users_list,10); self.clampsel(len(us))
|
|
scr.addstr(top+1,2,f"{'USER':<12}{'UID':>5} SUDO LIVE LOCK KEY LASTLOGIN (↑↓ · L lock · Uu unlock)",curses.color_pair(1))
|
|
y=top+2
|
|
for i,u in enumerate(us):
|
|
mark="»" if i==self.sel else " "
|
|
line=(f"{mark}{u['user']:<12}{u['uid']:>5} {'yes' if u['sudo'] else ' - ':<4} {u['live']:>3} "
|
|
f"{'L' if u['locked'] else '-'} {'y' if u['key'] else 'n'} {u['lastlogin']}")
|
|
scr.addstr(y,2,line[:w-4],curses.color_pair(2) if i==self.sel else (curses.color_pair(4) if u['locked'] else curses.color_pair(6))); y+=1
|
|
if self.sel<len(us):
|
|
u=us[self.sel]; scr.addstr(bot-1,2,f"groups: {u['groups']}"[:w-4],curses.color_pair(5))
|
|
|
|
def b_security(self,scr,top,bot,w):
|
|
banned=self.data("ban",f2b_banned_all,4); bl=self.data("blc2",blacklist_list,8)
|
|
scr.addstr(top+1,2,f"fail2ban banned: {len(banned)} permanent blacklist: {len(bl)} "
|
|
"(b ban · B blacklist · u unban · w whois · Enter detail · g blacklist view)",curses.color_pair(1))
|
|
scr.addstr(top+3,2,"TOP ATTACKERS (attempts · country):",curses.color_pair(1))
|
|
rows=self.sec_rows(); self.clampsel(len(rows)); y=top+4
|
|
for i,(_,o) in enumerate(rows[:bot-top-5]):
|
|
mark="»" if i==self.sel else " "
|
|
bnd="●BL" if (subprocess.run(["ipset","test","avni_blacklist",o["ip"]],capture_output=True).returncode==0) else ("●b" if any(o["ip"] in f2b_status(j)["ips"] for j in ["sshd"]) else " ")
|
|
line=f"{mark}{o['count']:>7} {o['ip']:<16}{geo(o['ip']):<14}{bnd:<4}"
|
|
scr.addstr(y,2,line[:w-34],curses.color_pair(2) if i==self.sel else curses.color_pair(6))
|
|
scr.addstr(y,min(w-30,2+len(line)),bar(o["count"],o["mx"],24),curses.color_pair(4)); y+=1
|
|
|
|
def footer(self):
|
|
s=SECTIONS[self.sec]
|
|
return {"Security":"b ban·B blacklist·u unban·w whois·Enter detail·r·q",
|
|
"Services":"s start·x stop·R restart·l logs·r·q",
|
|
"SSH":"k kick·K kick+blacklist·r·q","Users":"L lock·U unlock·r·q",
|
|
"Domains":"Enter detail·r·q","Traffic":"↑↓ window·r·q"}.get(s,"r refresh·q quit")
|
|
|
|
def detail_ip(self,ip):
|
|
d=ip_intel(ip)
|
|
lines=[f"IP {d['ip']} ({d['country']})",
|
|
f"attempts {d['attempts']}",
|
|
f"first {d['first']}", f"last {d['last']}",
|
|
f"banned {d['banned']} blacklisted {d['blacklisted']}",
|
|
f"org {d['org']}",
|
|
"usernames "+", ".join(f"{u}({c})" for u,c in d["users"]) or "usernames —","",
|
|
"[b]an temp [B]lacklist permanent [u]nban [w]hois [any] close"]
|
|
ch=self.popup(f"IP intel · {ip}",lines,wait=True)
|
|
if ch in (ord('b'),): self.msg="ban "+ip+": "+(ban(ip) or "ok"); self.cache.pop("ban",None)
|
|
elif ch==ord('B'): self.msg=blacklist_add(ip); self.cache.pop("blc2",None)
|
|
elif ch==ord('u'): self.msg="unban "+ip+": "+(unban(ip) or "ok"); self.cache.pop("ban",None)
|
|
elif ch==ord('w'): self.popup("whois "+ip,whois_ip(ip).splitlines(),wait=True)
|
|
|
|
def act(self,ch):
|
|
s=SECTIONS[self.sec]; rows=self.rows(); obj=rows[self.sel][1] if rows and self.sel<len(rows) else None
|
|
if s=="Traffic" and ch in (curses.KEY_UP,curses.KEY_DOWN,ord('j'),ord('k')):
|
|
self.tw+=1 if ch in (curses.KEY_DOWN,ord('j')) else -1; return
|
|
if s=="Security":
|
|
o=self.sec_rows()[self.sel][1] if self.sec_rows() and self.sel<len(self.sec_rows()) else None
|
|
ip=o["ip"] if o else None
|
|
if ch in (curses.KEY_ENTER,10,13) and ip: self.detail_ip(ip)
|
|
elif ch==ord('b') and ip: self.msg="ban "+ip+": "+(ban(ip) or "ok"); self.cache.pop("ban",None)
|
|
elif ch==ord('B') and ip: self.msg=blacklist_add(ip); self.cache.pop("blc2",None); self.cache.pop("blc",None)
|
|
elif ch==ord('u') and ip: self.msg="unban "+ip+": "+(unban(ip) or "ok"); self.cache.pop("ban",None)
|
|
elif ch==ord('w') and ip: self.popup("whois "+ip,whois_ip(ip).splitlines(),wait=True)
|
|
elif ch==ord('g'): self.popup("Permanent blacklist",[x+" "+geo(x.split('/')[0]) for x in blacklist_list()] or ["(empty)"],wait=True)
|
|
elif ch==ord('a'):
|
|
v=self.prompt("ban/blacklist IP or domain: ")
|
|
if v: self.msg=blacklist_add(v); self.cache.pop("blc2",None)
|
|
elif s=="Services" and obj:
|
|
n=obj["name"]
|
|
if ch==ord('R'): self.msg=f"restarting {n}…"; sh(f"docker restart {n}",40); self.cache.pop("svc",None); self.msg=f"restarted {n}"
|
|
elif ch==ord('x'): sh(f"docker stop {n}",40); self.cache.pop("svc",None); self.msg=f"stopped {n}"
|
|
elif ch==ord('s'): sh(f"docker start {n}",40); self.cache.pop("svc",None); self.msg=f"started {n}"
|
|
elif ch==ord('l'): self.popup("logs "+n,sh(f"docker logs --tail 60 {n}",10).splitlines(),wait=True)
|
|
elif s=="SSH" and obj:
|
|
if ch==ord('k'): self.msg=kick_tty(obj["tty"]); self.cache.pop("ssh",None)
|
|
elif ch==ord('K'): self.msg=kick_ip(obj["from"])+" + "+blacklist_add(obj["from"]); self.cache.pop("ssh",None)
|
|
elif s=="Domains" and obj and ch in (curses.KEY_ENTER,10,13):
|
|
d=obj["domain"]; code=sh(f"curl -s -m8 -o /dev/null -w %{{http_code}} https://{d}/")
|
|
self.popup("domain · "+d,[f"backend {obj['cont']}",f"HTTP {code}",f"cert exp {cert_expiry(d)}",
|
|
f"IPs {sh('getent ahostsv4 '+d).splitlines()[0].split()[0] if sh('getent ahostsv4 '+d) else '?'}"],wait=True)
|
|
elif s=="Users" and obj:
|
|
if ch==ord('L') and obj["user"] not in("root",): sh(f"usermod -L {obj['user']}"); self.cache.pop("usr",None); self.msg=f"locked {obj['user']}"
|
|
elif ch in (ord('U'),ord('u')): sh(f"usermod -U {obj['user']}"); self.cache.pop("usr",None); self.msg=f"unlocked {obj['user']}"
|
|
|
|
def prompt(self,label):
|
|
curses.echo(); curses.curs_set(1); h,w=self.scr.getmaxyx()
|
|
self.scr.addstr(h-1,0," "*(w-1),curses.color_pair(2)); self.scr.addstr(h-1,1,label,curses.color_pair(2))
|
|
try: v=self.scr.getstr(h-1,1+len(label),50).decode().strip()
|
|
except Exception: v=""
|
|
curses.noecho(); curses.curs_set(0); return v
|
|
def popup(self,title,lines,wait=False):
|
|
h,w=self.scr.getmaxyx(); win=curses.newwin(min(h-4,len(lines)+4),w-6,2,3); win.box()
|
|
win.addstr(0,2,f" {title} ",curses.color_pair(1))
|
|
for i,l in enumerate(lines[:h-7]):
|
|
try: win.addstr(1+i,2,str(l)[:w-10],curses.color_pair(6))
|
|
except Exception: pass
|
|
win.refresh(); ch=win.getch() if wait else -1; return ch
|
|
|
|
def loop(self):
|
|
curses.curs_set(0); self.scr.timeout(1500)
|
|
while True:
|
|
self.draw(); ch=self.scr.getch()
|
|
if ch==-1: continue
|
|
if ch in (ord('q'),27): break
|
|
elif ch in (curses.KEY_RIGHT,9): self.sec=(self.sec+1)%len(SECTIONS); self.sel=0
|
|
elif ch==curses.KEY_LEFT: self.sec=(self.sec-1)%len(SECTIONS); self.sel=0
|
|
elif ord('1')<=ch<=ord('7'): self.sec=ch-ord('1'); self.sel=0
|
|
elif ch in (curses.KEY_DOWN,) and SECTIONS[self.sec]!="Traffic": self.sel+=1
|
|
elif ch in (curses.KEY_UP,) and SECTIONS[self.sec]!="Traffic": self.sel=max(0,self.sel-1)
|
|
elif ch==ord('r'): self.cache.clear(); self.msg="refreshed"
|
|
else: self.act(ch)
|
|
|
|
def svc_line(r):
|
|
return f"{'●' if r['status'].startswith('Up') else '○'} {r['name']:<32}{r['status']:<24}{(' '+r['domain']) if r['domain'] else ''}"
|
|
def color_for(line):
|
|
if line.startswith("○") or "unhealthy" in line: return curses.color_pair(4)
|
|
if line.startswith("●"): return curses.color_pair(3)
|
|
return curses.color_pair(6)
|
|
def tui_main(scr): init_colors(); TUI(scr).loop()
|
|
|
|
def main():
|
|
if len(sys.argv)>1: cli(sys.argv[1:])
|
|
elif not sys.stdout.isatty(): print("avni: not a TTY. Try a subcommand (avni status) or run in a terminal."); sys.exit(1)
|
|
else: curses.wrapper(tui_main)
|
|
|
|
if __name__=="__main__": main()
|