#!/usr/bin/env python3
# Автор: Команда FiloSoft — https://filo-soft.ru
"""Restricted NDJSON SSH bridge; runs as pbxmon and connects to local AMI."""
import collections
import json
import os
import queue
import re
import select
import socket
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path

VERSION='1.1.0'
CONFIG=Path('/etc/pbx-monitor/config.json')
OUTPUT_LOCK=threading.Lock()
ALLOWED_ACTIONS={'Login','Logoff','Ping','CoreShowChannels','CoreStatus','CoreSettings','QueueStatus','SIPpeers','SIPshowregistry'}
FIELDS={'Event','Channel','ChannelState','ChannelStateDesc','CallerIDNum','CallerIDName','ConnectedLineNum','Context','Exten','Uniqueid','Linkedid','BridgeId','BridgeUniqueid','Duration','Application','ApplicationData','Queue','MemberName','Name','Location','Interface','Status','Paused','PausedReason','Penalty','CallsTaken','LastCall','InCall','Ringinuse','Position','Count','HoldTime','TalkTime','Reason','Cause','Cause-txt','DialStatus','DestChannel','DestUniqueid','DestLinkedid','Peer','PeerStatus','Address','IPaddress','IPport','StateInterface','Membership','Calls','Holdtime','Talktime','Completed','Abandoned','ServiceLevelPerf','Max','Strategy','AsteriskVersion','CoreStartupDate','CoreStartupTime','CoreCurrentCalls','CoreProcessedCalls','CoreReloadDate','CoreReloadTime'}
FIELDS.update({'UniqueID','LinkedID','StartTime','AnswerTime','EndTime','Source','Destination','DestinationContext','LastApplication','Disposition','BillableSeconds','RTT','ReportCount'})
LOG_EVENTS={'Newchannel','Newstate','Hangup','DialBegin','DialEnd','BridgeEnter','BridgeLeave','QueueCallerJoin','QueueCallerLeave','QueueCallerAbandon','AgentCalled','AgentConnect','AgentComplete','AgentRingNoAnswer','QueueMemberStatus','QueueMemberPaused','PeerStatus','Shutdown','Cdr','RTCPReceived'}

def emit(value):
    value['ts']=time.time()
    with OUTPUT_LOCK:sys.stdout.write(json.dumps(value,ensure_ascii=False)+'\n');sys.stdout.flush()

def sanitized(message):
    result={k:str(v)[:1500] for k,v in message.items() if (k in FIELDS or re.fullmatch(r'Report\d+FractionLost',k))and k!='ApplicationData'}
    if message.get('Application')=='Queue':result['ApplicationData']=str(message.get('ApplicationData','')).split(',')[0][:80]
    if message.get('Event')=='Cdr':
        for src,dst in [('StartTime','started'),('AnswerTime','answered'),('EndTime','ended')]:
            try:result[dst]=datetime.strptime(message[src],'%Y-%m-%d %H:%M:%S').astimezone().timestamp()
            except (KeyError,ValueError):pass
    return result

class AMI:
    def __init__(self,config):
        self.config=config;self.socket=None;self.pending={};self.lock=threading.Lock();self.send_lock=threading.Lock();self.channels={};self.ready=False;self.error=None;self.queue_cache={};self.cache_lock=threading.Lock()
        threading.Thread(target=self.run,daemon=True).start()
    def request(self,action,**values):
        if action not in ALLOWED_ACTIONS:raise ValueError('AMI action is not allowed')
        if self.socket is None:raise RuntimeError('AMI недоступен')
        request_id='pbx-'+str(time.time_ns());response_queue=queue.Queue()
        with self.lock:self.pending[request_id]=response_queue
        text='Action: '+action+'\r\nActionID: '+request_id+'\r\n'+''.join(k+': '+str(v)+'\r\n' for k,v in values.items())+'\r\n'
        result=[]
        try:
            with self.send_lock:self.socket.sendall(text.encode())
            deadline=time.monotonic()+8
            while time.monotonic()<deadline:
                try:message=response_queue.get(timeout=max(.05,deadline-time.monotonic()))
                except queue.Empty:raise RuntimeError('AMI response timed out')
                result.append(message)
                if message.get('Response')=='Error':raise RuntimeError('AMI rejected '+action)
                if message.get('Event','').endswith('Complete') or message.get('EventList')=='Complete':break
                if 'Response'in message and message.get('EventList')!='start' and action not in ['CoreShowChannels','QueueStatus','SIPpeers','SIPshowregistry']:break
            return result
        finally:
            with self.lock:self.pending.pop(request_id,None)
    def run(self):
        while True:
            try:
                s=socket.create_connection((self.config.get('ami_host','127.0.0.1'),int(self.config.get('ami_port',5038))),timeout=5);s.settimeout(1);self.socket=s
                reader=threading.Thread(target=self.read,args=(s,),daemon=True);reader.start()
                self.request('Login',Username=self.config['ami_user'],Secret=self.config['ami_secret'],Events='on')
                self.ready=True;self.error=None;emit({'type':'connection','ami':True})
                while reader.is_alive():reader.join(timeout=1)
            except Exception:
                self.error='Нет связи с локальным AMI'
            finally:
                self.ready=False
                if self.socket:
                    try:self.socket.close()
                    except OSError:pass
                self.socket=None;emit({'type':'connection','ami':False,'error':self.error or 'AMI соединение прервано'});time.sleep(3)
    def read(self,s):
        buffer=b''
        while True:
            try:data=s.recv(65536)
            except socket.timeout:continue
            except OSError:break
            if not data:break
            buffer+=data
            if len(buffer)>4*1024*1024:break
            while b'\r\n\r\n'in buffer:
                frame,buffer=buffer.split(b'\r\n\r\n',1)
                message={}
                for line in frame.decode(errors='replace').splitlines():
                    if ': 'in line:
                        key,value=line.split(': ',1);message[key]=value
                if not message:continue
                request_id=message.get('ActionID')
                with self.lock:pending=self.pending.get(request_id)
                if pending:pending.put(message)
                event=message.get('Event')
                if event in ['Newchannel','Newstate']:
                    key=message.get('Channel')
                    if key:self.channels[key]={**self.channels.get(key,{}),**sanitized(message)}
                    emit({'type':'channels','channels':list(self.channels.values())})
                elif event=='Hangup':
                    self.channels.pop(message.get('Channel'),None);emit({'type':'channels','channels':list(self.channels.values())})
                if event in LOG_EVENTS and not request_id:emit({'type':'event','data':sanitized(message)})

CPU_PREVIOUS=None
def metrics():
    global CPU_PREVIOUS
    result={'cpu':None,'memory':None,'disk':None,'uptime':None,'load':None}
    try:
        values=[int(x) for x in Path('/proc/stat').read_text().splitlines()[0].split()[1:]]
        total=sum(values[:8]);idle=values[3]+values[4]
        if CPU_PREVIOUS and total>CPU_PREVIOUS[0]:result['cpu']=round(100*(1-(idle-CPU_PREVIOUS[1])/(total-CPU_PREVIOUS[0])),2)
        CPU_PREVIOUS=(total,idle)
        mem={line.split(':')[0]:int(line.split()[1]) for line in Path('/proc/meminfo').read_text().splitlines()}
        result['memory']=round(100*(1-mem['MemAvailable']/mem['MemTotal']),2)
        result['memory_total_mb']=round(mem['MemTotal']/1024)
        result['uptime']=float(Path('/proc/uptime').read_text().split()[0]);result['load']=os.getloadavg()[0]
        disk=os.statvfs('/');result['disk']=round(100*(1-disk.f_bavail/disk.f_blocks),2)
    except (OSError,ValueError,KeyError):pass
    for service in ['asterisk','fail2ban']:
        try:result[service]=subprocess.run(['systemctl','is-active',service],capture_output=True,text=True,timeout=3).stdout.strip()=='active'
        except (OSError,subprocess.TimeoutExpired):result[service]=None
    try:
        p=subprocess.run(['/usr/sbin/asterisk','-rx','odbc show'],capture_output=True,text=True,timeout=4)
        connected=re.findall(r'Connected:\s*(Yes|No)',p.stdout,re.I)
        result['odbc']=all(x.lower()=='yes' for x in connected) if connected else None
    except (OSError,subprocess.TimeoutExpired):result['odbc']=None
    return result

def f2b(request):
    context=dict(request);connection=os.environ.get('SSH_CONNECTION','').split()
    if connection:context['client_ip']=connection[0]
    p=subprocess.run(['sudo','-n','/usr/local/sbin/pbx-monitor-f2b'],input=json.dumps(context)+'\n',capture_output=True,text=True,timeout=18)
    try:result=json.loads(p.stdout)
    except ValueError:return {'ok':False,'error':'Нет доступа к Fail2Ban helper'}
    return result

def state_source(request):
    p=subprocess.run(['sudo','-n','/usr/local/sbin/pbx-monitor-state'],input=json.dumps(request)+'\n',capture_output=True,text=True,timeout=14)
    try:result=json.loads(p.stdout)
    except ValueError:raise RuntimeError('Нет доступа к источнику статистики')from None
    if not result.get('ok'):raise RuntimeError(result.get('error','Источник статистики недоступен'))
    return result['data']

def queue_query(config,name):
    # A Realtime QueueStatus may query thousands of records when unfiltered.
    # Use a named queue on an isolated connection; core telemetry stays responsive.
    s=socket.create_connection((config['ami_host'],config['ami_port']),timeout=3);s.settimeout(.5);buffer=b''
    def request(action,values,seconds):
        nonlocal buffer
        request_id='queue-'+str(time.time_ns());messages=[]
        text='Action: '+action+'\r\nActionID: '+request_id+'\r\n'+''.join(k+': '+str(v)+'\r\n'for k,v in values.items())+'\r\n'
        s.sendall(text.encode());deadline=time.monotonic()+seconds
        while time.monotonic()<deadline:
            try:chunk=s.recv(65536)
            except socket.timeout:continue
            if not chunk:raise RuntimeError('AMI очередь: соединение прервано')
            buffer+=chunk
            if len(buffer)>4*1024*1024:raise RuntimeError('AMI очередь: ответ слишком большой')
            while b'\r\n\r\n'in buffer:
                frame,buffer=buffer.split(b'\r\n\r\n',1)
                m=dict(line.split(': ',1)for line in frame.decode(errors='replace').splitlines()if ': 'in line)
                if m.get('ActionID')!=request_id:continue
                if m.get('Response')=='Error':raise RuntimeError('AMI отклонил запрос очереди')
                messages.append(m)
                if m.get('Event','').endswith('Complete')or(action=='Login'and m.get('Response')=='Success'):return messages
        raise RuntimeError('Очередь не ответила вовремя. Проверьте Realtime/ODBC и номер очереди')
    try:
        request('Login',{'Username':config['ami_user'],'Secret':config['ami_secret'],'Events':'off'},3)
        messages=request('QueueStatus',{'Queue':name},22)
        return [sanitized(x)for x in messages if x.get('Event')in ['QueueParams','QueueMember','QueueEntry']]
    finally:s.close()

def queue_response(config,ami,request_id,name):
    try:
        events=queue_query(config,name)
        with ami.cache_lock:ami.queue_cache[name]=events
        emit({'type':'response','id':request_id,'ok':True,'data':{'name':name,'events':events}})
    except Exception as e:emit({'type':'response','id':request_id,'ok':False,'error':str(e)if isinstance(e,RuntimeError)else 'Не удалось прочитать очередь'})

def log_tail():
    paths=['/var/log/asterisk/messages.log','/var/log/asterisk/security.log']
    result=[]
    for value in paths:
        try:
            with Path(value).open('rb') as f:
                f.seek(0,2);size=f.tell();f.seek(max(0,size-50000));lines=f.read().decode(errors='replace').splitlines()[-60:]
            for line in lines:
                if re.search(r'password|secret|authorization|DBPASS|DTMF|PrivateKey|(?:Executing|System).*\b(?:Set|System|SHELL)\b',line,re.I):line='[Скрыта строка с чувствительными данными]'
                result.append({'source':Path(value).name,'text':line[:2500]})
        except OSError:pass
    return result[-100:]

def main():
    config=json.loads(CONFIG.read_text());ami=AMI(config)
    emit({'type':'hello','version':VERSION,'capabilities':['channels','queues','events','logs','metrics','fail2ban']})
    for raw in sys.stdin:
        request_id=None
        try:
            if len(raw)>32768:raise ValueError('Request is too large')
            request=json.loads(raw);request_id=request.get('id');op=request.get('op')
            if op=='snapshot':
                ready=ami.ready
                channel_messages=ami.request('CoreShowChannels') if ready else []
                channels=[sanitized(x) for x in channel_messages if x.get('Event')=='CoreShowChannel']
                # Retain linkedid learned from events when an older snapshot omits it.
                channels=[{**ami.channels.get(x.get('Channel'),{}),**x} for x in channels]
                if ready:ami.channels={x['Channel']:x for x in channels if x.get('Channel')}
                with ami.cache_lock:queues=[x for value in ami.queue_cache.values()for x in value]
                settings=ami.request('CoreSettings') if ready else []
                version=next((x.get('AsteriskVersion') for x in settings if x.get('AsteriskVersion')),None)
                f=f2b({'op':'snapshot'})
                system=metrics()
                try:system.update(state_source({'op':'health'}))
                except RuntimeError:pass
                data={'ami':ready,'version':version,'channels':channels if ready else None,'queues':queues,'metrics':system,'fail2ban':f.get('data') if f.get('ok') else {'available':False,'error':f.get('error')}}
            elif op=='queue':
                name=str(request.get('name',''))
                if not re.fullmatch(r'[A-Za-z0-9_.:@-]{1,80}',name):raise ValueError('Invalid queue name')
                threading.Thread(target=queue_response,args=(config,ami,request_id,name),daemon=True).start()
                continue
            elif op=='f2b':
                result=f2b(request.get('request',{}))
                if not result.get('ok'):raise RuntimeError(result.get('error','Fail2Ban request failed'))
                data=result['data']
            elif op=='logs':data={'lines':log_tail()}
            elif op=='records':data=state_source({'op':'records','checkpoint':request.get('checkpoint')or {}})
            elif op=='ping':data={'version':VERSION,'ami':ami.ready}
            else:raise ValueError('Operation is not allowed')
            emit({'type':'response','id':request_id,'ok':True,'data':data})
        except Exception as e:
            message=str(e) if isinstance(e,(ValueError,RuntimeError)) else 'Ошибка запроса к локальному мосту'
            emit({'type':'response','id':request_id,'ok':False,'error':message})

if __name__=='__main__':
    try:main()
    except (BrokenPipeError,KeyboardInterrupt):pass