"""Authenticated LAN coordination service; intentionally has no shell endpoint."""
import argparse
import hmac
import ipaddress
import json
import mimetypes
import os
import re
import stat
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, unquote, urlsplit

from .store import Conflict, Invalid, Store

WEB = Path(__file__).resolve().parents[1] / 'assets' / 'dashboard' / 'dist'


def restricted_token(path):
    path = Path(path)
    flags = os.O_RDONLY | getattr(os, 'O_NOFOLLOW', 0)
    fd = os.open(str(path), flags)
    try:
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or (os.name == 'posix' and (info.st_uid != os.getuid() or info.st_mode & 0o077)):
            raise ValueError('token file must be owned by the service user and mode 600')
        with os.fdopen(fd, 'r', encoding='utf-8') as stream:
            fd = None
            value = stream.read(4097).strip()
        if len(value) < 32 or len(value) > 4096 or any(c.isspace() for c in value):
            raise ValueError('token must contain 32 to 4096 non-whitespace characters')
        return value
    finally:
        if fd is not None:
            os.close(fd)


def local_runtime(path):
    root = Path(path).expanduser().resolve()
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    if os.name == 'posix':
        info = root.stat()
        if info.st_uid != os.getuid() or info.st_mode & 0o077:
            raise ValueError('runtime root must be owned by service user and mode 700')
        # Longest mount-point match handles nested mounts and bind mounts.
        mounts = []
        for line in Path('/proc/mounts').read_text().splitlines():
            fields = line.split()
            mount = fields[1].replace('\\040', ' ').replace('\\134', '\\')
            if str(root) == mount or str(root).startswith(mount.rstrip('/') + '/'):
                mounts.append((len(mount), fields[2]))
        if not mounts or max(mounts)[1] not in {'xfs','ext4','ext3','ext2','btrfs','tmpfs','overlay','f2fs','zfs'}:
            raise ValueError('runtime root must use a verified local filesystem')
    elif str(root).startswith('\\\\'):
        raise ValueError('runtime root must be local')
    return root


class MetricsCache:
    def __init__(self, config):
        self.config = config
        self.value = {'status':'waiting', 'message':'等待首次采集'}
        self.lock = threading.Lock()
        self.stop = threading.Event()
        self.thread = threading.Thread(target=self.run, daemon=True)
        self.thread.start()

    def run(self):
        while not self.stop.is_set():
            try:
                from .metrics import collect
                value = collect(self.config)
                # Reject non-JSON collectors before exposing the cache.
                json.dumps(value, allow_nan=False)
            except Exception:
                value = {'status':'unavailable', 'message':'采集暂不可用'}
            with self.lock:
                self.value = {'sampled_at':time.time(), 'data':value}
            self.stop.wait(15)

    def snapshot(self):
        with self.lock:
            return self.value


class EtaClock:
    def __init__(self, store, interval=15):
        self.store = store
        self.interval = interval
        self.stop = threading.Event()
        self.status = {'status':'waiting','last_tick':None}
        self.thread = threading.Thread(target=self.run, name='eta-clock', daemon=True)
        self.thread.start()

    def run(self):
        while not self.stop.is_set():
            try:
                self.store.eta_tick()
                self.status = {'status':'ok','last_tick':time.time()}
            except Exception:
                self.status = {'status':'unavailable','last_tick':self.status['last_tick']}
            self.stop.wait(self.interval)

    def close(self):
        self.stop.set()
        self.thread.join(timeout=16)


class Server(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, config):
        bind = config['bind']
        address = ipaddress.IPv4Address(bind)
        allowed = any(address in ipaddress.ip_network(net) for net in ('10.0.0.0/8','172.16.0.0/12','192.168.0.0/16','127.0.0.0/8'))
        if not allowed:
            raise ValueError('bind must be an explicit LAN or loopback IPv4')
        port = config.get('port', 8796)
        if isinstance(port, bool) or not isinstance(port, int) or not 0 <= port <= 65535:
            raise ValueError('invalid port')
        self.primary_session_id = config.get('primary_session_id', 'human-progress')
        if not isinstance(self.primary_session_id, str) or not self.primary_session_id or len(self.primary_session_id) > 128:
            raise ValueError('invalid primary_session_id')
        self.token = restricted_token(Path(config['token_file']).expanduser())
        self.web_root = Path(config.get('web_root', WEB)).expanduser().resolve()
        self.title = config.get('title', 'Agent Workbench')
        if not isinstance(self.title,str) or not 1 <= len(self.title) <= 200:
            raise ValueError('invalid title')
        root = local_runtime(config['runtime_root'])
        dbpath = root / 'control.sqlite3'
        if dbpath.is_symlink():
            raise ValueError('database cannot be a symlink')
        self.store = Store(dbpath)
        if os.name == 'posix':
            dbpath.chmod(0o600)
        super().__init__((bind, port), Handler)
        self.authority = bind + ':' + str(self.server_port)
        self.allowed_hosts = set(config.get('allowed_hosts', [])) | {self.authority}
        if any(not isinstance(host,str) or not host or '/' in host or '@' in host or any(c.isspace() for c in host) for host in self.allowed_hosts):
            raise ValueError('invalid allowed_hosts')
        self.metrics = MetricsCache(config.get('metrics', {}))
        self.eta_clock = EtaClock(self.store)

    def server_close(self):
        if hasattr(self, 'eta_clock'):
            self.eta_clock.close()
        if hasattr(self, 'metrics'):
            self.metrics.stop.set()
        super().server_close()


class Handler(BaseHTTPRequestHandler):
    server_version = 'AgentControl'
    sys_version = ''

    def setup(self):
        super().setup()
        self.connection.settimeout(10)

    def log_message(self, *args):
        # Do not log request targets, bodies, headers or credentials.
        pass

    def output(self, code, value, content_type='application/json; charset=utf-8'):
        payload = value if isinstance(value, bytes) else json.dumps(value, ensure_ascii=False, allow_nan=False).encode('utf-8')
        self.send_response(code)
        self.send_header('Content-Type', content_type)
        self.send_header('Content-Length', str(len(payload)))
        self.send_header('Cache-Control', 'no-store')
        self.send_header('X-Content-Type-Options', 'nosniff')
        self.send_header('Referrer-Policy', 'no-referrer')
        self.send_header('Content-Security-Policy', "default-src 'self'; script-src 'self'; style-src 'self'; style-src-attr 'unsafe-inline'; connect-src 'self'; object-src 'none'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
        self.end_headers()
        self.wfile.write(payload)

    def guard(self, write=False):
        hosts = self.headers.get_all('Host', [])
        if len(hosts) != 1 or hosts[0] not in self.server.allowed_hosts:
            self.output(403, {'error':'host rejected'})
            return False
        origins = self.headers.get_all('Origin', [])
        if origins and origins != ['http://' + hosts[0]]:
            self.output(403, {'error':'origin rejected'})
            return False
        public_navigation = (
            not write and self.command == 'GET'
            and not urlsplit(self.path).path.startswith('/api/')
            and self.headers.get('Sec-Fetch-Mode') == 'navigate'
            and self.headers.get('Sec-Fetch-Dest') == 'document')
        if self.headers.get('Sec-Fetch-Site') == 'cross-site' and not public_navigation:
            self.output(403, {'error':'cross-site request rejected'})
            return False
        if write and self.headers.get('X-CSRF-Token') != '1':
            self.output(403, {'error':'csrf rejected'})
            return False
        return True

    def auth(self):
        supplied = self.headers.get('Authorization', '')
        if not hmac.compare_digest(supplied.encode('utf-8'), ('Bearer ' + self.server.token).encode('utf-8')):
            self.output(401, {'error':'authentication required'})
            return False
        return True

    def do_GET(self):
        if not self.guard():
            return
        parts = urlsplit(self.path)
        if parts.path.startswith('/api/'):
            if not self.auth():
                return
            try:
                if parts.path == '/api/state' and not parts.query:
                    state = self.server.store.state()
                    state['title'] = self.server.title
                    state['primary_session_id'] = self.server.primary_session_id
                    state['metrics'] = self.server.metrics.snapshot()
                    state['eta_clock'] = dict(self.server.eta_clock.status)
                    self.output(200, state)
                elif parts.path == '/api/queue-health' and not parts.query:
                    self.output(200,self.server.store.queue_health())
                elif parts.path == '/api/task-reports':
                    query = parse_qs(parts.query,strict_parsing=True,keep_blank_values=True)
                    if set(query) != {'task_id'} or len(query['task_id']) != 1:
                        raise Invalid('invalid report query')
                    self.output(200,self.server.store.task_reports(query['task_id'][0]))
                elif parts.path == '/api/logs':
                    query = parse_qs(parts.query,strict_parsing=True,keep_blank_values=True) if parts.query else {}
                    if set(query)-{'level','status'} or any(len(v)!=1 for v in query.values()):
                        raise Invalid('invalid log query')
                    self.output(200,self.server.store.logs(
                        level=query.get('level',[None])[0],status=query.get('status',[None])[0]))
                elif parts.path == '/api/eta-notifications' and not parts.query:
                    self.output(200, self.server.store.eta_notifications())
                elif parts.path == '/api/decision-notifications' and not parts.query:
                    self.output(200, self.server.store.decision_notifications())
                elif parts.path == '/api/hook-notifications' and not parts.query:
                    self.output(200, self.server.store.hook_notifications())
                elif parts.path in ('/api/messages','/api/conversation'):
                    query = parse_qs(parts.query, strict_parsing=True)
                    if set(query) != {'session_id'} or len(query['session_id']) != 1:
                        raise Invalid('invalid query')
                    reader = self.server.store.conversation if parts.path == '/api/conversation' else self.server.store.messages
                    self.output(200, {'messages':reader(query['session_id'][0])})
                else:
                    self.output(404, {'error':'not found'})
            except (Invalid,ValueError):
                self.output(400, {'error':'invalid request'})
            except Exception:
                self.output(500, {'error':'request failed'})
            return
        # Only built dashboard files under the configured root are public.
        if parts.query:
            self.output(404, {'error':'not found'})
            return
        relative = unquote(parts.path).lstrip('/') or 'index.html'
        allowed_suffixes = {'.html','.js','.css','.svg','.ico','.png','.webp','.woff','.woff2'}
        target = (self.server.web_root / relative).resolve()
        if ('\\' in relative or any(p in ('.','..') for p in relative.split('/'))
                or not target.is_relative_to(self.server.web_root) or target.suffix not in allowed_suffixes):
            self.output(404, {'error':'not found'})
            return
        if not target.is_file():
            if relative == 'index.html' and not self.server.web_root.is_dir():
                self.output(503, {'error':'dashboard not built; run the documented frontend build'})
                return
            self.output(404, {'error':'not found'})
            return
        content_type = mimetypes.guess_type(str(target))[0] or 'application/octet-stream'
        if target.suffix in ('.html','.js','.css','.svg'):
            content_type += '; charset=utf-8'
        self.output(200, target.read_bytes(), content_type)

    def do_POST(self):
        self.mutate()

    def do_PATCH(self):
        self.mutate()

    def mutate(self):
        if not self.guard(write=True) or not self.auth():
            return
        try:
            if self.headers.get('Content-Type', '').split(';')[0] != 'application/json' or self.headers.get('Transfer-Encoding'):
                raise Invalid('invalid content type')
            lengths = self.headers.get_all('Content-Length', [])
            if len(lengths) != 1:
                raise Invalid('length required')
            length = int(lengths[0])
            body_limit = 4 * 1024 * 1024 if self.path in ('/api/mirror','/api/worker/complete') else 65536
            if not 0 < length <= body_limit:
                raise Invalid('invalid length')
            data = json.loads(self.rfile.read(length))
            if not isinstance(data, dict):
                raise Invalid('object required')
            store = self.server.store
            routes = {'/api/task-reports':store.task_report, '/api/attention':store.attention, '/api/eta-notifications/ack':store.eta_ack, '/api/decisions':store.decision, '/api/decision-notifications/ack':store.decision_ack,
                      '/api/hook-notifications/submitting':store.hook_submitting,
                      '/api/hook-notifications/coalesce':store.hook_coalesce, '/api/hooks':store.hook, '/api/hooks/escalate':store.hook_escalate, '/api/hooks/ack':store.hook_ack,
                      '/api/sessions':store.session, '/api/session-requests':store.request,
                      '/api/tasks':store.task, '/api/messages':store.send,
                      '/api/worker/claim':store.claim, '/api/worker/renew':store.renew, '/api/worker/complete':store.complete,
                      '/api/events':store.event, '/api/mirror':store.mirror, '/api/worker/start':store.start,
                      '/api/worker/submission':store.submission, '/api/worker/uncertain':store.uncertain, '/api/messages/reconcile':store.reconcile}
            if self.command == 'PATCH' and self.path.startswith('/api/tasks/'):
                result = store.task(data, self.path[len('/api/tasks/'):])
            elif self.command == 'POST' and re.fullmatch(r'/api/tasks/[A-Za-z0-9_.:-]+/integrate',self.path):
                result = store.integrate(self.path.split('/')[3],data)
            elif self.command == 'POST' and re.fullmatch(r'/api/eta-reviews/[A-Za-z0-9_.:-]+/complete',self.path):
                result = store.eta_complete(self.path.split('/')[3],data)
            elif self.command == 'POST' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+/respond',self.path):
                result = store.decision_respond(self.path.split('/')[3],data)
            elif self.command == 'POST' and self.path in routes:
                result = routes[self.path](data)
            else:
                self.output(404, {'error':'not found'})
                return
            self.output(200, result)
        except Conflict:
            self.output(409, {'error':'state conflict'})
        except (Invalid, ValueError, TypeError, KeyError):
            self.output(400, {'error':'invalid request'})
        except Exception:
            self.output(500, {'error':'request failed'})


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--config', required=True)
    args = parser.parse_args()
    config = json.loads(Path(args.config).read_text(encoding='utf-8-sig'))
    os.umask(0o077)
    server = Server(config)
    try:
        print('Agent control listening at http://' + server.authority, flush=True)
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()


if __name__ == '__main__':
    main()