#!/usr/bin/env python3 import argparse import datetime as dt import hashlib import hmac import json import os import re import secrets import sqlite3 import subprocess import tempfile import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import pymysql import bcrypt UTC = dt.timezone.utc PUBLIC_KEY_RE = re.compile(r'^ssh-ed25519 [A-Za-z0-9+/=]+(?: .{0,200})?$') IDENTIFIER_RE = re.compile(r'^[A-Za-z0-9_]+$') def now(): return dt.datetime.now(UTC) def iso(value): return value.astimezone(UTC).isoformat().replace('+00:00', 'Z') class Config: def __init__(self): self.bind_host = os.environ.get('BOOTSTRAP_BIND_HOST', '127.0.0.1') self.port = int(os.environ.get('BOOTSTRAP_PORT', '4092')) self.state_db = Path(os.environ.get('BOOTSTRAP_STATE_DB', '/var/lib/cangku-bootstrap/state.sqlite3')) self.token_pepper = os.environ['BOOTSTRAP_TOKEN_PEPPER'].encode() self.mysql_host = os.environ.get('BOOTSTRAP_MYSQL_HOST', '127.0.0.1') self.mysql_port = int(os.environ.get('BOOTSTRAP_MYSQL_PORT', '3307')) self.mysql_database = os.environ.get('BOOTSTRAP_MYSQL_DATABASE', 'cangku_system') self.mysql_user = os.environ['BOOTSTRAP_MYSQL_USER'] self.mysql_password = os.environ['BOOTSTRAP_MYSQL_PASSWORD'] self.ssh_ca_key = Path(os.environ.get('BOOTSTRAP_SSH_CA_KEY', '/var/lib/cangku-bootstrap/ssh_user_ca')) self.ssh_host_key = Path(os.environ.get('BOOTSTRAP_SSH_HOST_KEY', '/etc/ssh/ssh_host_ed25519_key.pub')) self.ssh_host = os.environ['BOOTSTRAP_SSH_HOST'] self.ssh_port = int(os.environ.get('BOOTSTRAP_SSH_PORT', '22')) self.ssh_user = os.environ.get('BOOTSTRAP_SSH_USER', 'cangku-tunnel') self.ssh_remote_db_host = os.environ.get('BOOTSTRAP_REMOTE_DB_HOST', '127.0.0.1') self.ssh_remote_db_port = int(os.environ.get('BOOTSTRAP_REMOTE_DB_PORT', '3307')) self.cert_days = int(os.environ.get('BOOTSTRAP_CERT_DAYS', '30')) if not IDENTIFIER_RE.fullmatch(self.mysql_database): raise ValueError('BOOTSTRAP_MYSQL_DATABASE contains an invalid identifier') class Broker: def __init__(self, config): self.config = config self.lock = threading.Lock() config.state_db.parent.mkdir(parents=True, exist_ok=True) self._initialize_state() def _connect_state(self): connection = sqlite3.connect(self.config.state_db, timeout=15) connection.row_factory = sqlite3.Row return connection def _initialize_state(self): with self._connect_state() as connection: connection.executescript(''' CREATE TABLE IF NOT EXISTS enrollment_tokens ( id TEXT PRIMARY KEY, label TEXT NOT NULL, token_hash TEXT NOT NULL UNIQUE, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, max_uses INTEGER NOT NULL, use_count INTEGER NOT NULL DEFAULT 0, revoked_at TEXT ); CREATE TABLE IF NOT EXISTS installations ( id TEXT PRIMARY KEY, token_id TEXT NOT NULL, label TEXT NOT NULL, db_username TEXT NOT NULL UNIQUE, certificate_serial INTEGER NOT NULL UNIQUE, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, revoked_at TEXT, FOREIGN KEY(token_id) REFERENCES enrollment_tokens(id) ); CREATE TABLE IF NOT EXISTS counters ( name TEXT PRIMARY KEY, value INTEGER NOT NULL ); INSERT OR IGNORE INTO counters(name, value) VALUES ('certificate_serial', 1000); ''') columns = { row['name'] for row in connection.execute('PRAGMA table_info(installations)') } if 'admin_username' not in columns: connection.execute('ALTER TABLE installations ADD COLUMN admin_username TEXT') def token_hash(self, token): return hmac.new(self.config.token_pepper, token.encode(), hashlib.sha256).hexdigest() def create_token(self, label, max_uses, days): token_id = secrets.token_hex(8) token = f'ck_live_{token_id}_{secrets.token_urlsafe(32)}' created = now() expires = created + dt.timedelta(days=days) with self._connect_state() as connection: connection.execute( 'INSERT INTO enrollment_tokens(id,label,token_hash,created_at,expires_at,max_uses) VALUES(?,?,?,?,?,?)', (token_id, label, self.token_hash(token), iso(created), iso(expires), max_uses), ) return token, token_id, expires def _find_token(self, connection, token): row = connection.execute( 'SELECT * FROM enrollment_tokens WHERE token_hash=?', (self.token_hash(token),) ).fetchone() if not row or row['revoked_at']: raise PermissionError('部署 Key 无效或已停用') if dt.datetime.fromisoformat(row['expires_at'].replace('Z', '+00:00')) <= now(): raise PermissionError('部署 Key 已过期') if row['use_count'] >= row['max_uses']: raise PermissionError('部署 Key 已达到使用次数上限') return row def _next_serial(self, connection): connection.execute("UPDATE counters SET value=value+1 WHERE name='certificate_serial'") return connection.execute("SELECT value FROM counters WHERE name='certificate_serial'").fetchone()[0] def _mysql_connection(self): return pymysql.connect( host=self.config.mysql_host, port=self.config.mysql_port, user=self.config.mysql_user, password=self.config.mysql_password, charset='utf8mb4', autocommit=True, connect_timeout=8, ) def _create_database_user(self, username, password): database = self.config.mysql_database with self._mysql_connection() as connection: with connection.cursor() as cursor: cursor.execute(f"CREATE USER '{username}'@'%%' IDENTIFIED BY %s", (password,)) cursor.execute(f"GRANT ALL PRIVILEGES ON `{database}`.* TO '{username}'@'%'") def _drop_database_user(self, username): if not re.fullmatch(r'cangku_[a-f0-9]{16}', username): raise ValueError('Refusing to drop an unexpected database username') with self._mysql_connection() as connection: with connection.cursor() as cursor: cursor.execute(f"DROP USER IF EXISTS '{username}'@'%'") def _create_application_admin(self, username, password, label): password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=12)).decode() email = f'{username}@remote.local' display_name = f'远程开发者 {label}'[:50] with self._mysql_connection() as connection: with connection.cursor() as cursor: cursor.execute( f'''INSERT INTO `{self.config.mysql_database}`.users (username,password_hash,name,email,role,status,auth_source) VALUES(%s,%s,%s,%s,'admin','active','local')''', (username, password_hash, display_name, email), ) def _remove_application_admin(self, username): with self._mysql_connection() as connection: with connection.cursor() as cursor: cursor.execute( f'DELETE FROM `{self.config.mysql_database}`.users WHERE username=%s', (username,), ) def _deactivate_application_admin(self, username): if not username: return with self._mysql_connection() as connection: with connection.cursor() as cursor: cursor.execute( f"UPDATE `{self.config.mysql_database}`.users SET status='inactive' WHERE username=%s", (username,), ) def _sign_public_key(self, public_key, installation_id, serial, expires): if len(public_key) > 1024 or not PUBLIC_KEY_RE.fullmatch(public_key): raise ValueError('仅接受有效的 Ed25519 SSH 公钥') with tempfile.TemporaryDirectory(dir=self.config.state_db.parent) as directory: public_key_path = Path(directory) / 'client.pub' public_key_path.write_text(public_key + '\n', encoding='utf-8') subprocess.run( [ 'ssh-keygen', '-q', '-s', str(self.config.ssh_ca_key), '-I', installation_id, '-z', str(serial), '-n', self.config.ssh_user, '-V', f'-5m:+{self.config.cert_days}d', str(public_key_path), ], check=True, capture_output=True, text=True, timeout=10, ) return public_key_path.with_name('client-cert.pub').read_text(encoding='utf-8').strip() def enroll(self, token, label, public_key): label = str(label or '').strip() if not label or len(label) > 80 or any(ord(char) < 32 for char in label): raise ValueError('设备名称必须为 1 到 80 个可见字符') with self.lock: installation_id = secrets.token_hex(16) db_username = f'cangku_{secrets.token_hex(8)}' db_password = secrets.token_urlsafe(32) admin_username = f'dev_{installation_id[:12]}' admin_password = secrets.token_urlsafe(18) created = now() expires = created + dt.timedelta(days=self.config.cert_days) db_created = False admin_created = False with self._connect_state() as connection: connection.execute('BEGIN IMMEDIATE') token_row = self._find_token(connection, token) serial = self._next_serial(connection) try: certificate = self._sign_public_key(public_key, installation_id, serial, expires) self._create_database_user(db_username, db_password) db_created = True self._create_application_admin(admin_username, admin_password, label) admin_created = True connection.execute( '''INSERT INTO installations (id,token_id,label,db_username,certificate_serial,created_at,expires_at,admin_username) VALUES(?,?,?,?,?,?,?,?)''', (installation_id, token_row['id'], label, db_username, serial, iso(created), iso(expires), admin_username), ) connection.execute( 'UPDATE enrollment_tokens SET use_count=use_count+1 WHERE id=?', (token_row['id'],) ) connection.commit() except Exception: connection.rollback() if admin_created: self._remove_application_admin(admin_username) if db_created: self._drop_database_user(db_username) raise return { 'installation_id': installation_id, 'ssh': { 'host': self.config.ssh_host, 'port': self.config.ssh_port, 'user': self.config.ssh_user, 'certificate': certificate, 'host_key': self.config.ssh_host_key.read_text(encoding='utf-8').strip(), 'valid_until': iso(expires), }, 'database': { 'remote_host': self.config.ssh_remote_db_host, 'remote_port': self.config.ssh_remote_db_port, 'name': self.config.mysql_database, 'username': db_username, 'password': db_password, }, 'application_admin': { 'username': admin_username, 'password': admin_password, }, } def list_state(self): with self._connect_state() as connection: tokens = [dict(row) for row in connection.execute( 'SELECT id,label,created_at,expires_at,max_uses,use_count,revoked_at FROM enrollment_tokens ORDER BY created_at DESC' )] installations = [dict(row) for row in connection.execute( 'SELECT id,token_id,label,db_username,admin_username,certificate_serial,created_at,expires_at,revoked_at FROM installations ORDER BY created_at DESC' )] return {'tokens': tokens, 'installations': installations} def revoke_installation(self, installation_id): with self.lock, self._connect_state() as connection: row = connection.execute('SELECT * FROM installations WHERE id=?', (installation_id,)).fetchone() if not row: raise ValueError('未找到设备') if not row['revoked_at']: self._drop_database_user(row['db_username']) self._deactivate_application_admin(row['admin_username']) connection.execute('UPDATE installations SET revoked_at=? WHERE id=?', (iso(now()), installation_id)) def revoke_token(self, token_id): with self._connect_state() as connection: result = connection.execute( 'UPDATE enrollment_tokens SET revoked_at=? WHERE id=? AND revoked_at IS NULL', (iso(now()), token_id) ) if result.rowcount != 1: raise ValueError('未找到可停用的 Key') class Handler(BaseHTTPRequestHandler): server_version = 'CangkuBootstrap/1.0' def send_json(self, status, value): body = json.dumps(value, ensure_ascii=False).encode() self.send_response(status) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.send_header('Cache-Control', 'no-store') self.send_header('X-Content-Type-Options', 'nosniff') self.end_headers() self.wfile.write(body) def do_GET(self): if self.path == '/health': self.send_json(200, {'status': 'ok'}) else: self.send_json(404, {'error': 'not found'}) def do_POST(self): if self.path != '/v1/enroll': self.send_json(404, {'error': 'not found'}) return try: length = int(self.headers.get('Content-Length', '0')) if length <= 0 or length > 16 * 1024: raise ValueError('请求大小无效') authorization = self.headers.get('Authorization', '') if not authorization.startswith('Bearer '): raise PermissionError('缺少部署 Key') payload = json.loads(self.rfile.read(length)) result = self.server.broker.enroll( authorization.removeprefix('Bearer ').strip(), payload.get('label'), payload.get('public_key', '').strip(), ) self.send_json(201, result) except PermissionError as error: self.send_json(401, {'error': str(error)}) except (ValueError, json.JSONDecodeError) as error: self.send_json(400, {'error': str(error)}) except Exception as error: print(f'enrollment failed: {type(error).__name__}: {error}', flush=True) self.send_json(500, {'error': '注册服务内部错误'}) def log_message(self, format_string, *args): client = self.headers.get('X-Real-IP') or self.client_address[0] print(f'{client} - {format_string % args}', flush=True) def build_parser(): parser = argparse.ArgumentParser(description='Cangku remote deployment enrollment broker') subparsers = parser.add_subparsers(dest='command', required=True) subparsers.add_parser('serve') create = subparsers.add_parser('create-key') create.add_argument('--label', required=True) create.add_argument('--max-uses', type=int, default=1) create.add_argument('--days', type=int, default=30) subparsers.add_parser('list') revoke = subparsers.add_parser('revoke-installation') revoke.add_argument('installation_id') revoke_key = subparsers.add_parser('revoke-key') revoke_key.add_argument('token_id') return parser def main(): args = build_parser().parse_args() config = Config() broker = Broker(config) if args.command == 'serve': server = ThreadingHTTPServer((config.bind_host, config.port), Handler) server.broker = broker server.serve_forever() elif args.command == 'create-key': if args.max_uses < 1 or args.max_uses > 1000 or args.days < 1 or args.days > 365: raise SystemExit('max-uses must be 1..1000 and days must be 1..365') token, token_id, expires = broker.create_token(args.label, args.max_uses, args.days) print(json.dumps({'id': token_id, 'key': token, 'expires_at': iso(expires), 'max_uses': args.max_uses})) elif args.command == 'list': print(json.dumps(broker.list_state(), ensure_ascii=False, indent=2)) elif args.command == 'revoke-installation': broker.revoke_installation(args.installation_id) elif args.command == 'revoke-key': broker.revoke_token(args.token_id) if __name__ == '__main__': main()