#!/usr/bin/env python3 # Автор: Команда FiloSoft — https://filo-soft.ru """Root-owned helper. Only asterisk jail and typed IP operations are allowed.""" import ast import fcntl import ipaddress import json import os import re import subprocess import sys import tempfile import time from pathlib import Path JAIL='asterisk' OVERLAY=Path('/etc/fail2ban/jail.d/99-pbx-monitor.local') LOCK=Path('/run/lock/pbx-monitor-f2b.lock') MARKER='# Managed by PBX Monitor' def command(*args): p=subprocess.run(['/usr/bin/fail2ban-client',*args],capture_output=True,text=True,timeout=10,env={**os.environ,'LC_ALL':'C'}) if p.returncode:raise RuntimeError('Fail2Ban rejected the request or asterisk jail is unavailable') return p.stdout.strip() def address(value,network=False): if not isinstance(value,str) or len(value)>80 or any(c.isspace() for c in value):raise ValueError('Invalid IP address') if network: parsed=ipaddress.ip_network(value,strict=False) if parsed.prefixlen==0:raise ValueError('A whole Internet subnet is not allowed') return str(parsed) if '/'in value else str(ipaddress.ip_address(value)) return str(ipaddress.ip_address(value)) def entries(raw): try: parsed=ast.literal_eval(raw) if isinstance(parsed,(tuple,list)):return [str(x) for x in parsed] except (ValueError,SyntaxError):pass result=[] for value in re.split(r'[\s,\[\]\'"|`]+',raw): try: norm=str(ipaddress.ip_network(value,strict=False)) if '/'in value else str(ipaddress.ip_address(value)) if norm not in result:result.append(norm) except ValueError:continue return result def snapshot(): try: status=command('status',JAIL) banned=entries(command('get',JAIL,'banip')) ignored=entries(command('get',JAIL,'ignoreip')) metrics={} for name,label in [('failed_current','Currently failed'),('failed_total','Total failed'),('banned_current','Currently banned'),('banned_total','Total banned')]: m=re.search(re.escape(label)+r':\s*(\d+)',status) metrics[name]=int(m[1]) if m else None return {'available':True,'jail':JAIL,'banned':banned,'ignoreip':ignored,'metrics':metrics,'bantime':command('get',JAIL,'bantime'),'persistent_overlay':OVERLAY.exists()} except (OSError,RuntimeError,subprocess.TimeoutExpired): return {'available':False,'jail':JAIL,'banned':None,'ignoreip':None,'error':'Fail2Ban или jail asterisk недоступен'} def protected(target,context): subnet=ipaddress.ip_network(target,strict=False) protected_ips=['127.0.0.1','::1'] client=context.get('client_ip') if client: try:protected_ips.append(str(ipaddress.ip_address(client))) except ValueError:pass for value in protected_ips: p=ipaddress.ip_address(value) if p.version==subnet.version and p in subnet:return True return subnet.is_loopback or subnet.is_link_local or subnet.is_multicast or subnet.is_unspecified def atomic_write(path,content): path.parent.mkdir(parents=True,exist_ok=True) fd,tmp=tempfile.mkstemp(prefix='.pbx-monitor-',dir=path.parent) try: with os.fdopen(fd,'w') as f:f.write(content);f.flush();os.fsync(f.fileno()) os.chmod(tmp,0o640);os.replace(tmp,path) finally: if os.path.exists(tmp):os.unlink(tmp) def update_ignore(op,target): current=entries(command('get',JAIL,'ignoreip')) before=OVERLAY.read_text() if OVERLAY.exists() else None if before is not None and not before.startswith(MARKER):raise RuntimeError('Existing overlay is not owned by PBX Monitor') desired=list(current) if op=='add-ignore' and target not in desired:desired.append(target) if op=='del-ignore':desired=[x for x in desired if x!=target] if desired==current and (OVERLAY.exists() or op=='del-ignore'):return {'changed':False,'persistent':OVERLAY.exists()} # Preserve all effective existing entries; no other jail is edited. text=MARKER+'\n[asterisk]\nignoreip = '+' '.join(desired)+'\n' backup=OVERLAY.with_suffix('.local.backup') if before is not None:atomic_write(backup,before) atomic_write(OVERLAY,text) applied=False try: command('-t') command('set',JAIL,'addignoreip' if op=='add-ignore' else 'delignoreip',target) applied=True after=entries(command('get',JAIL,'ignoreip')) if set(after)!=set(desired):raise RuntimeError('Runtime and persistent ignoreip differ') return {'changed':True,'persistent':True} except Exception: if before is None:OVERLAY.unlink(missing_ok=True) else:atomic_write(OVERLAY,before) if applied: try:command('set',JAIL,'delignoreip' if op=='add-ignore' else 'addignoreip',target) except Exception:pass raise def execute(request): op=request.get('op') if op=='snapshot':return snapshot() if op not in ['ban','unban','add-ignore','del-ignore']:raise ValueError('Operation is not allowed') target=address(request.get('ip'),network=op.endswith('ignore')) if op=='add-ignore' and request.get('unban') and '/'in target:raise ValueError('Unban a single IP separately from a CIDR whitelist entry') if op in ['ban','del-ignore'] and protected(target,request):raise ValueError('A loopback or management address is protected') if op=='ban': existing=entries(command('get',JAIL,'banip')) if target not in existing:command('set',JAIL,'banip',target) confirmed=entries(command('get',JAIL,'banip')) if target not in confirmed:raise RuntimeError('Ban could not be confirmed') result={'changed':target not in existing} elif op=='unban': existing=entries(command('get',JAIL,'banip')) if target in existing:command('set',JAIL,'unbanip',target) if target in entries(command('get',JAIL,'banip')):raise RuntimeError('Unban could not be confirmed') result={'changed':target in existing} else: result=update_ignore(op,target) if op=='add-ignore' and request.get('unban'): if '/'not in target:command('set',JAIL,'unbanip',target) else:raise ValueError('Unban a single IP separately from a CIDR whitelist entry') return {**result,'operation':op,'ip':target,'snapshot':snapshot()} def main(): try: raw=sys.stdin.readline(16385) if len(raw)>16384:raise ValueError('Request is too large') request=json.loads(raw) LOCK.parent.mkdir(parents=True,exist_ok=True) with LOCK.open('w') as lock: fcntl.flock(lock,fcntl.LOCK_EX) result=execute(request) print(json.dumps({'ok':True,'data':result},ensure_ascii=False)) except Exception as e: message=str(e) if isinstance(e,(ValueError,RuntimeError)) else 'Fail2Ban operation failed' print(json.dumps({'ok':False,'error':message},ensure_ascii=False));sys.exit(1) if __name__=='__main__':main()