#!/usr/bin/env python3 """Publish authoritative conversation read state with automatic rollback.""" import base64 import fcntl import hashlib import hmac import json import os import pwd import shutil 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 = 'becf85bcf60f087ef6b4e52d8c55b3e47f930fb95dcf710739715ed402bf93ab' BASE = Path('/www/server/xingyu-im/releases/read-badge-20260901-183130') ALLOWED = {'backend/internal/app/im.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.read-badge-' + 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 verify_read_state(conn, values): selected = query(conn, """SELECT m.user_id,p.nickname,security.token_version,m.conversation_id,m.read_seq,c.last_seq, (SELECT COUNT(*) FROM im_messages unread_msg WHERE unread_msg.conversation_id=m.conversation_id AND unread_msg.seq>GREATEST(m.read_seq,m.clear_seq,m.join_seq) AND unread_msg.sender_id<>m.user_id AND unread_msg.recalled_at IS NULL AND unread_msg.admin_removed_at IS NULL) FROM im_conversation_members m JOIN im_conversations c ON c.id=m.conversation_id JOIN users u ON u.id=m.user_id JOIN user_profiles p ON p.user_id=u.id JOIN user_security_controls security ON security.user_id=u.id WHERE u.is_test=1 AND u.status=1 AND u.deleted_at IS NULL AND m.status=1 ORDER BY m.user_id,m.conversation_id LIMIT 1""") assert selected, 'no demonstration conversation available for read-state verification' user_id, nickname, version, conversation_id, read_seq, last_seq, unread = selected[0] token = user_token(values['IM_JWT_SECRET'], user_id, nickname, version) body = json.dumps({'readSeq': read_seq}).encode() request = urllib.request.Request( DOMAIN + '/api/v1/im/conversations/' + str(conversation_id) + '/settings', data=body, method='PATCH', headers={'Authorization': 'Bearer ' + token, 'Accept': 'application/json', 'Content-Type': 'application/json'}, ) with urllib.request.urlopen(request, timeout=20) as response: assert response.status == 200 data = json.loads(response.read())['data'] assert data['success'] is True expected_ack_seq = last_seq if int(unread) == 0 and int(last_seq) > int(read_seq) else read_seq assert int(data['readSeq']) == int(expected_ack_seq) assert int(data['unread']) == int(unread) return {'userId': user_id, 'conversationId': conversation_id, 'storedReadSeq': read_seq, 'ackReadSeq': expected_ack_seq, 'unread': unread} 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/read-badge-' + 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: migrations = dict(query(conn, 'SELECT version,checksum FROM schema_migrations')) for name, sha in migrations.items(): assert new_source['backend/migrations/' + name] == sha assert len(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-read-badge-' + stamp) backup.mkdir(mode=0o700) shutil.copy2(str(BT / 'xingyu-api'), str(backup / 'xingyu-api-original')) state = { 'backup': str(backup), 'stage': str(stage), 'previousBackendSha256': OLD, 'backendSha256': new_hash, 'archiveSha256': expected, 'changedSourceFiles': sorted(changed), 'frontendBefore': frontend, 'databaseMigrationsApplied': [], 'businessDataWrites': False, } (backup / 'before.json').write_text(json.dumps(state, indent=2)) print('BACKUP_VERIFIED=' + str(backup), flush=True) changed_binary = 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')) == migrations assert {name: str((WEB / name).resolve()) for name in frontend} == frontend for path in ['/api/v1/im/conversations', '/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['readState'] = verify_read_state(conn, values) state['healthAndAuthChecks'] = True (backup / 'result.json').write_text(json.dumps(state, indent=2)) print(json.dumps(state), flush=True) except Exception: 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 was required.\n') raise finally: conn.close() if __name__ == '__main__': publish(*sys.argv[1:])