#!/usr/bin/env python3 """Publish authoritative realtime presence with automatic rollback.""" import base64 import fcntl import hashlib import hmac import json import os import pwd import shutil import socket import ssl import sys import time import urllib.error import urllib.request import zipfile from pathlib import Path sys.path.insert(0, '/www/server/xingyu-im/ops') from baota_publish_test_users import BT, DOMAIN, WEB, database, digest, env_values, extract, fetch, health, query, restart OLD = 'e8e071ec7006a32303f7cd654cddad8afda22ecb4fe83464305a9d3aef4d7280' BASE = Path('/www/server/xingyu-im/releases/nearby-post-media-20260901-115131') ALLOWED = { 'backend/internal/app/auth.go', 'backend/internal/app/public_auth.go', 'backend/internal/app/im.go', 'backend/internal/app/social.go', 'backend/internal/app/presence_test.go', 'backend/internal/testusers/seed.go', } def source_hashes(archive): with zipfile.ZipFile(str(archive)) as package: result = {} for item in package.infolist(): if item.is_dir(): continue name = item.filename.replace('\\', '/') assert name not in result and not name.startswith('/') and '..' not in Path(name).parts result[name] = hashlib.sha256(package.read(item)).hexdigest() return result def running(expected): pids = [] for proc in Path('/proc').iterdir(): if not proc.name.isdigit(): continue try: if os.readlink(str(proc / 'exe')) == str(BT / 'xingyu-api'): assert digest(proc / 'exe') == expected pids.append(int(proc.name)) except (FileNotFoundError, PermissionError): pass assert len(pids) == 1, 'expected one running project executable' return pids def replace(source, stamp): temporary = BT / ('xingyu-api.presence-' + stamp) assert not temporary.exists() shutil.copy2(str(source), str(temporary)) www = pwd.getpwnam('www') os.chown(str(temporary), www.pw_uid, www.pw_gid) os.chmod(str(temporary), 0o750) os.replace(str(temporary), str(BT / 'xingyu-api')) def encode_json(value): raw = json.dumps(value, separators=(',', ':'), ensure_ascii=False).encode() return base64.urlsafe_b64encode(raw).rstrip(b'=').decode() def user_token(secret, user_id, nickname, version): now = int(time.time()) header = encode_json({'alg': 'HS256', 'typ': 'JWT'}) payload = encode_json({'role': 'user', 'name': nickname, 'ver': version, 'sub': str(user_id), 'iat': now, 'exp': now + 300}) signing_input = (header + '.' + payload).encode() signature = base64.urlsafe_b64encode(hmac.new(secret.encode(), signing_input, hashlib.sha256).digest()).rstrip(b'=').decode() return header + '.' + payload + '.' + signature def receive_exact(stream, count): output = b'' while len(output) < count: chunk = stream.recv(count - len(output)) if not chunk: raise RuntimeError('WebSocket closed before AUTH_ACK') output += chunk return output def first_websocket_command(stream, initial=b''): class Buffered: def __init__(self, buffered, source): self.buffered, self.source = buffered, source def recv(self, count): if self.buffered: chunk, self.buffered = self.buffered[:count], self.buffered[count:] return chunk return self.source.recv(count) reader = Buffered(initial, stream) first, second = receive_exact(reader, 2) size = second & 0x7f if size == 126: size = int.from_bytes(receive_exact(reader, 2), 'big') elif size == 127: size = int.from_bytes(receive_exact(reader, 8), 'big') return json.loads(receive_exact(reader, size)).get('command') def profile_online(user_id, token): request = urllib.request.Request( DOMAIN + '/api/v1/users/' + str(user_id), headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json'}, ) with urllib.request.urlopen(request, timeout=20) as response: assert response.status == 200 return bool(json.loads(response.read())['data']['online']) def verify_presence(conn, values): selected = query(conn, """SELECT target.id,target_profile.nickname,target_security.token_version, viewer.id,viewer_profile.nickname,viewer_security.token_version FROM users target JOIN user_profiles target_profile ON target_profile.user_id=target.id JOIN user_security_controls target_security ON target_security.user_id=target.id JOIN user_privacy_settings target_privacy ON target_privacy.user_id=target.id JOIN users viewer ON viewer.id<>target.id AND viewer.status=1 AND viewer.deleted_at IS NULL JOIN user_profiles viewer_profile ON viewer_profile.user_id=viewer.id JOIN user_security_controls viewer_security ON viewer_security.user_id=viewer.id WHERE target.status=1 AND target.deleted_at IS NULL AND target.is_test=1 AND target_privacy.online_visible=1 AND NOT EXISTS(SELECT 1 FROM user_blocks block WHERE (block.user_id=viewer.id AND block.blocked_user_id=target.id) OR (block.user_id=target.id AND block.blocked_user_id=viewer.id)) ORDER BY target.id,viewer.id LIMIT 1""") assert selected, 'no isolated presence verification users available' target_id, target_name, target_version, viewer_id, viewer_name, viewer_version = selected[0] target_token = user_token(values['IM_JWT_SECRET'], target_id, target_name, target_version) viewer_token = user_token(values['IM_JWT_SECRET'], viewer_id, viewer_name, viewer_version) key = base64.b64encode(os.urandom(16)).decode() request = ('\r\n'.join([ 'GET /ws HTTP/1.1', 'Host: im.bchongw.com', 'Upgrade: websocket', 'Connection: Upgrade', 'Sec-WebSocket-Version: 13', 'Sec-WebSocket-Key: ' + key, 'Origin: https://im.bchongw.com', 'Sec-WebSocket-Protocol: xingyu.jwt.' + target_token, ]) + '\r\n\r\n').encode() context = ssl.create_default_context() with socket.create_connection(('im.bchongw.com', 443), timeout=10) as raw: with context.wrap_socket(raw, server_hostname='im.bchongw.com') as stream: stream.settimeout(10) stream.sendall(request) response = b'' while b'\r\n\r\n' not in response: response += stream.recv(4096) head, remainder = response.split(b'\r\n\r\n', 1) assert b' 101 ' in head.split(b'\r\n', 1)[0] assert first_websocket_command(stream, remainder) == 'AUTH_ACK' assert profile_online(target_id, viewer_token), 'connected user was reported offline' disconnected = False for _ in range(30): if not profile_online(target_id, viewer_token): disconnected = True break time.sleep(0.1) assert disconnected, 'closed final socket was still reported online' return {'targetUserId': target_id, 'viewerUserId': viewer_id, 'connected': True, 'disconnected': True} def publish(archive, expected): assert os.geteuid() == 0 archive = Path(archive).resolve() assert archive.parent == Path('/tmp') and digest(archive) == expected lock = open('/www/server/xingyu-im/ops/application-release.lock', 'a') fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) assert digest(BT / 'xingyu-api') == OLD, 'production changed; re-audit required' running(OLD) health() stamp = time.strftime('%Y%m%d-%H%M%S') stage = Path('/www/server/xingyu-im/releases/presence-' + stamp) extract(archive, stage) os.chmod(str(stage), 0o700) old_source = source_hashes(BASE / 'backend-source.zip') new_source = source_hashes(stage / 'backend-source.zip') changed = {name for name in set(old_source) | set(new_source) if old_source.get(name) != new_source.get(name)} assert changed == ALLOWED, 'unreviewed source changes: ' + repr(changed) new_hash = digest(stage / 'xingyu-api') assert (stage / 'xingyu-api').read_bytes()[:4] == b'\x7fELF' and new_hash != OLD manifest = json.loads((stage / 'manifest.json').read_text()) for name in ['xingyu-api', 'backend-source.zip']: assert digest(stage / name) == manifest['files'][name] values = env_values() conn, _ = database(values) try: before_migrations = dict(query(conn, 'SELECT version,checksum FROM schema_migrations')) for name, sha in before_migrations.items(): assert new_source['backend/migrations/' + name] == sha assert len(before_migrations) == len([name for name in new_source if name.startswith('backend/migrations/')]) frontend = {name: str((WEB / name).resolve()) for name in ['admin', 'app']} backup = Path('/www/backup/xingyu-presence-' + stamp) backup.mkdir(mode=0o700) shutil.copy2(str(BT / 'xingyu-api'), str(backup / 'xingyu-api-original')) privacy_before = query(conn, """SELECT privacy.user_id,privacy.online_visible FROM user_privacy_settings privacy JOIN users user ON user.id=privacy.user_id WHERE user.is_test=1 AND user.deleted_at IS NULL ORDER BY privacy.user_id""") assert privacy_before, 'expected existing demonstration accounts' privacy_backup = backup / 'privacy-before.json' privacy_backup.write_text(json.dumps(privacy_before)) os.chmod(str(privacy_backup), 0o600) state = { 'backup': str(backup), 'stage': str(stage), 'previousBackendSha256': OLD, 'backendSha256': new_hash, 'archiveSha256': expected, 'changedSourceFiles': sorted(changed), 'frontendUnchanged': frontend, 'databaseMigrationsApplied': [], 'businessDataWrites': True, 'privacyScope': 'existing is_test=1 accounts only', } (backup / 'before.json').write_text(json.dumps(state, indent=2)) print('BACKUP_VERIFIED=' + str(backup), flush=True) changed_binary = False privacy_changed = False try: replace(stage / 'xingyu-api', stamp) changed_binary = True restart() state['runningPids'] = running(new_hash) assert digest(BT / 'xingyu-api') == new_hash assert dict(query(conn, 'SELECT version,checksum FROM schema_migrations')) == before_migrations assert {name: str((WEB / name).resolve()) for name in frontend} == frontend with conn.cursor() as cursor: cursor.execute("""UPDATE user_privacy_settings privacy JOIN users user ON user.id=privacy.user_id SET privacy.online_visible=1 WHERE user.is_test=1 AND user.deleted_at IS NULL""") state['demonstrationPrivacyRowsMatched'] = cursor.rowcount privacy_changed = True for path in ['/api/v1/nearby/users', '/admin/v1/users', '/ws']: try: fetch(path) except urllib.error.HTTPError as error: assert error.code == 401 else: raise RuntimeError('private endpoint became public') state['presence'] = verify_presence(conn, values) state['healthAndAuthChecks'] = True (backup / 'result.json').write_text(json.dumps(state, indent=2)) print(json.dumps(state), flush=True) except Exception: if privacy_changed: with conn.cursor() as cursor: cursor.executemany('UPDATE user_privacy_settings SET online_visible=%s WHERE user_id=%s', [(online, user_id) for user_id, online in privacy_before]) if changed_binary: replace(backup / 'xingyu-api-original', stamp + '-rollback') restart() running(OLD) (backup / 'rolled-back.txt').write_text('Original binary restored. No database restore performed.\n') raise finally: conn.close() if __name__ == '__main__': publish(*sys.argv[1:])