388 lines
20 KiB
Python
388 lines
20 KiB
Python
"""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):
|
|
self.decision_scope = None
|
|
supplied = self.headers.get('Authorization', '')
|
|
if hmac.compare_digest(supplied.encode('utf-8'), ('Bearer ' + self.server.token).encode('utf-8')):
|
|
return True
|
|
scope = self.server.store.decision_credential_scope(supplied[7:]) if supplied.startswith('Bearer ') else None
|
|
if scope is None:
|
|
self.output(401, {'error':'authentication required'})
|
|
return False
|
|
path = urlsplit(self.path).path
|
|
base = '/api/decisions/' + scope
|
|
allowed = ((self.command=='GET' and path in (base,base+'/history',base+'/consultation'))
|
|
or (self.command=='PATCH' and path==base)
|
|
or (self.command=='POST' and path==base+'/facts'))
|
|
if not allowed or (self.command!='GET' and self.server.store.decision_get(scope)['status']=='completed'):
|
|
self.output(403, {'error':'credential scope rejected'})
|
|
return False
|
|
self.decision_scope = scope
|
|
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)
|
|
try:
|
|
entry = re.search(r'<script\b[^>]*\bsrc="(/assets/[^"/]+\.js)"', (self.server.web_root/'index.html').read_text(encoding='utf-8'))
|
|
state['frontend_entrypoint'] = entry.group(1) if entry else None
|
|
except OSError:
|
|
state['frontend_entrypoint'] = None
|
|
self.output(200, state)
|
|
elif parts.path == '/api/decision-bridge-bindings' and not parts.query:
|
|
self.output(200,self.server.store.decision_bindings())
|
|
elif re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+(?:/(?:history|consultation))?',parts.path) and not parts.query:
|
|
segments=parts.path.split('/')
|
|
method = {'history':self.server.store.decision_history,'consultation':self.server.store.decision_consultation}.get(segments[-1]) if len(segments)==5 else self.server.store.decision_get
|
|
self.output(200,method(segments[3]))
|
|
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 Conflict:
|
|
self.output(409, {'error':'state conflict'})
|
|
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')
|
|
if getattr(self,'decision_scope',None) and self.command=='PATCH':
|
|
allowed={'expected_revision','reason','title','summary','question','details','impact','recommendation','options','deadline','priority'}
|
|
if data.keys()-allowed:
|
|
self.output(403, {'error':'credential content scope rejected'})
|
|
return
|
|
store = self.server.store
|
|
routes = {'/api/decision-notifications/submitting':store.decision_submitting, '/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 == 'PATCH' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+',self.path):
|
|
result = store.decision_patch(self.path.split('/')[3],data)
|
|
elif self.command == 'POST' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+/facts/[A-Za-z0-9_.:-]+/answer',self.path):
|
|
segments=self.path.split('/')
|
|
result = store.decision_fact_answer(segments[3],segments[5],data)
|
|
elif self.command == 'POST' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+/(respond|review|complete|archive-receipt|bind-session|consultation|consultation-credential|facts)',self.path):
|
|
actions = {'respond':store.decision_respond,'review':store.decision_review,'complete':store.decision_complete,
|
|
'archive-receipt':store.decision_archive,'bind-session':store.decision_bind,
|
|
'consultation':store.decision_consultation,'consultation-credential':store.decision_credential,'facts':store.decision_fact}
|
|
if self.path.endswith('/bind-session') and data.get('session_id')==self.server.primary_session_id:
|
|
raise Conflict('primary session cannot be bound')
|
|
result = actions[self.path.split('/')[-1]](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()
|