#!/usr/bin/env python3 """Black-box HTTP/JSON regression checks against the recorded Go reference (stdlib only). compatibility_expected.txt holds the observable responses of the original Go server (v1.0.0), which the Apple and Android clients were built against. Any difference is a client-visible API change. After an intentional change, review the diff and re-record with --update. """ import argparse import ast import base64 import concurrent.futures import contextlib import http.client import hashlib import hmac import json import os import pprint from pathlib import Path import socket import sqlite3 import subprocess import tempfile import time ROOT = Path(__file__).resolve().parents[1] EXPECTED = Path(__file__).with_name('compatibility_expected.txt') SECRET = 'compatibility-test-secret-not-for-deployment-0123456789' PASSWORD = 'compatibility password' class Server: def __init__(self, binary, database, **extra): with socket.socket() as sock: sock.bind(('127.0.0.1', 0)) self.port = sock.getsockname()[1] env = os.environ.copy() env.update(BALANCE_ADDR=f'127.0.0.1:{self.port}', BALANCE_DB_PATH=str(database), BALANCE_JWT_SECRET=SECRET, BALANCE_ACCESS_TTL='15m', BALANCE_REFRESH_TTL='720h', BALANCE_ALLOW_REGISTRATION='true', BALANCE_CORS_ORIGINS='https://allowed.example') env.update(extra) self.log = tempfile.TemporaryFile() self.process = subprocess.Popen([str(binary)], env=env, cwd=ROOT, stdout=self.log, stderr=self.log) deadline = time.monotonic() + 20 while time.monotonic() < deadline: if self.process.poll() is not None: self.log.seek(0) raise AssertionError(f'{binary.name} exited: {self.log.read().decode()}') try: if self.request('GET', '/health')[0] == 200: return except (OSError, http.client.HTTPException): pass time.sleep(.05) self.close() raise AssertionError(f'{binary.name} startup timeout') def close(self): if self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout=5) except subprocess.TimeoutExpired: self.process.kill() self.process.wait() self.log.close() def request(self, method, path, body=None, token=None, headers=None, raw=None, decode_json=True): hdr = dict(headers or {}) if token: hdr['Authorization'] = f'Bearer {token}' if body is not None: raw = json.dumps(body).encode() hdr['Content-Type'] = 'application/json' connection = http.client.HTTPConnection('127.0.0.1', self.port, timeout=30) try: connection.request(method, path, body=raw, headers=hdr) response = connection.getresponse() data = response.read() values = response.getheaders() content_type = response.getheader('Content-Type', '') if decode_json and data and 'application/json' in content_type: data = json.loads(data) return response.status, values, data finally: connection.close() @contextlib.contextmanager def running(binary, database, **extra): server = Server(binary, database, **extra) try: yield server finally: server.close() def expect(result, status): assert result[0] == status, f'expected HTTP {status}, got {result[0]}: {result[2]!r}' return result[2] def credentials(email='User@Example.COM'): return {'email': email, 'password': PASSWORD} def change(identifier, stamp='2025-01-01T00:00:00Z', **extra): result = dict(entity='transaction', id=identifier, updatedAt=stamp, payload={'amount': 125.5, 'note': 'Тест €', 'nested': [None, True, 1]}) result.update(extra) return result def sync(server, token, changes=(), cursor=0, **extra): return expect(server.request('POST', '/v1/sync', dict(cursor=cursor, deviceId='compatibility', changes=list(changes), **extra), token), 200) def scenario(binary, database): """Return deterministic observable results; assertions check generated sessions too.""" observed = {} with running(binary, database) as server: for method, path, body in [ ('GET', '/v1/me', None), ('GET', '/v1/missing', None), ('POST', '/v1/auth/register', {'email': 'bad', 'password': PASSWORD}), ('POST', '/v1/auth/register', {'email': 'a@b', 'password': 'short'}), ('POST', '/v1/auth/login', credentials()), ('POST', '/v1/auth/refresh', {}), ('POST', '/v1/auth/refresh', {'refreshToken': 'bad'}), ('POST', '/v1/auth/login', {'unknown': 1}), ]: result = server.request(method, path, body) observed[f'{method} {path} {body}'] = (result[0], result[2]) for path in ['/','/app.js','/styles.css','/icon.svg','/manifest.webmanifest', '/service-worker.js','/missing-file']: result = server.request('GET', path) head = {k.lower(): v for k, v in result[1]} keys = ['content-type','cache-control','x-content-type-options', 'x-frame-options','referrer-policy','permissions-policy','content-security-policy'] observed[path] = (result[0], {k: head.get(k) for k in keys}, result[2]) assert server.request('HEAD', path)[2] == b'' for origin in ['https://allowed.example', 'https://denied.example']: result = server.request('OPTIONS', '/v1/sync', headers={'Origin': origin}) observed[origin] = (result[0], {k.lower(): v for k, v in result[1] if k.lower().startswith('access-control') or k.lower() == 'vary'}) session_result = server.request('POST', '/v1/auth/register', credentials(), headers={'X-Forwarded-Proto': 'https'}) session = expect(session_result, 201) assert session['user']['email'] == 'user@example.com' cookies = [v for k, v in session_result[1] if k.lower() == 'set-cookie'] assert len(cookies) == 2 for cookie in cookies: assert all(part in cookie for part in ['HttpOnly','Secure','SameSite=Strict','Path=/']) cookie_header = '; '.join(c.split(';')[0] for c in cookies) assert expect(server.request('GET', '/v1/me', headers={'Cookie': cookie_header}), 200) == session['user'] token = session['accessToken'] assert expect(server.request('GET', '/v1/me', token=token), 200) == session['user'] observed['invalid bearer'] = expect(server.request('GET', '/v1/me', token='bad'), 401) observed['duplicate'] = expect(server.request('POST', '/v1/auth/register', credentials()), 409) expect(server.request('POST', '/v1/auth/login', credentials()), 200) for payload in [{'cursor': -1}, {'extra': 1}, {'changes': [change('x', entity='unknown')]}, {'changes': [change('')]}, {'changes': [dict(entity='category', id='x')]}, {'changes': [change('x', payload=None)]}]: result = server.request('POST', '/v1/sync', payload, token) data = result[2] if isinstance(data, dict): data.pop('serverTime', None) observed[f'invalid sync {payload}'] = (result[0], data) observed['malformed'] = expect(server.request('POST', '/v1/sync', token=token, raw=b'{'), 400) first = sync(server, token, [change('one'), change('cat', entity='category'), change('budget', entity='budget')]) first.pop('serverTime') observed['first'] = first repeated = sync(server, token, [change('one'), change('one', '2024-01-01T00:00:00Z')], cursor=first['cursor']) repeated.pop('serverTime') observed['idempotent'] = repeated assert repeated['changes'] == [] and repeated['cursor'] == first['cursor'] deleted = sync(server, token, [change('one', '2025-01-02T00:00:00Z', deleted=True)], cursor=first['cursor']) deleted.pop('serverTime') observed['deleted'] = deleted assert deleted['changes'][0]['deleted'] and 'payload' not in deleted['changes'][0] pushed = sync(server, token, [change(f'page-{n}') for n in range(505)], cursor=deleted['cursor'], pull=False) assert pushed['cursor'] == deleted['cursor'] and pushed['changes'] == [] page = sync(server, token, cursor=deleted['cursor']) assert len(page['changes']) == 500 and page['hasMore'] tail = sync(server, token, cursor=page['cursor']) assert len(tail['changes']) == 5 and not tail['hasMore'] observed['pagination'] = (page['cursor'], tail['cursor']) with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda n: sync(server, token, [change(f'parallel-{n}')], pull=False), range(16))) assert all(result['changes'] == [] for result in results) parallel = sync(server, token, cursor=tail['cursor']) assert len(parallel['changes']) == 16 assert len({x['sequence'] for x in parallel['changes']}) == 16 other = expect(server.request('POST', '/v1/auth/register', credentials('other@example.com')), 201) assert sync(server, other['accessToken'])['changes'] == [] refreshed = expect(server.request('POST', '/v1/auth/refresh', {}, headers={'Cookie': cookie_header}), 200) assert refreshed['refreshToken'] != session['refreshToken'] observed['refresh replay'] = expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': session['refreshToken']}), 401) logged_out = server.request('POST', '/v1/auth/logout', {'refreshToken': refreshed['refreshToken']}) expect(logged_out, 204) assert len([v for k,v in logged_out[1] if k.lower() == 'set-cookie' and 'Max-Age=0' in v]) == 2 observed['logout refresh'] = expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': refreshed['refreshToken']}), 401) # Fresh limiter state permits exactly 30 authentication attempts, including invalid JSON. with running(binary, database) as server: statuses = [server.request('POST', '/v1/auth/login', raw=b'{')[0] for _ in range(31)] assert statuses == [400] * 30 + [429], statuses result = server.request('POST', '/v1/auth/login', raw=b'{') assert {k.lower():v for k,v in result[1]}.get('retry-after') == '60' observed['rate limit'] = result[2] with running(binary, database, BALANCE_ALLOW_REGISTRATION='false', BALANCE_ACCESS_TTL='1s') as server: observed['registration disabled'] = expect(server.request('POST', '/v1/auth/register', credentials()), 403) session = expect(server.request('POST', '/v1/auth/login', credentials()), 200) time.sleep(1.15) observed['expired access'] = expect(server.request('GET', '/v1/me', token=session['accessToken']), 401) return observed def takeover(first, second, database): with running(first, database) as server: session = expect(server.request('POST', '/v1/auth/register', credentials()), 201) original = sync(server, session['accessToken'], [change('persisted')]) with running(second, database) as server: assert expect(server.request('GET', '/v1/me', token=session['accessToken']), 200) == session['user'] assert expect(server.request('POST', '/v1/auth/login', credentials()), 200)['user'] == session['user'] assert sync(server, session['accessToken'])['changes'] == original['changes'] next_session = expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': session['refreshToken']}), 200) updated = sync(server, next_session['accessToken'], [change('persisted','2025-01-02T00:00:00Z',deleted=True)], cursor=original['cursor']) assert len(updated['changes']) == 1 and updated['changes'][0]['version'] == 2 with running(first, database) as server: assert expect(server.request('GET', '/v1/me', token=next_session['accessToken']), 200) == session['user'] expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': session['refreshToken']}), 401) expect(server.request('POST', '/v1/auth/refresh', {'refreshToken': next_session['refreshToken']}), 200) restored = sync(server, session['accessToken'], cursor=original['cursor']) assert restored['changes'] == updated['changes'] and restored['cursor'] == updated['cursor'] expect(server.request('POST', '/v1/auth/login', credentials()), 200) def edge_scenarios(binary, database): """Exercise decoder, routing, validation, range and session boundary behavior.""" observed = {} def record(name, result, headers=()): selected = {k.lower(): v for k, v in result[1] if k.lower() in headers} data = result[2] if isinstance(data, dict): data = {k: v for k, v in data.items() if k != 'serverTime'} observed[name] = (result[0], selected, data) with running(binary, database) as server: session = expect(server.request('POST', '/v1/auth/register', credentials()), 201) token = session['accessToken'] for method in ['POST', 'PUT', 'DELETE', 'PATCH', 'HEAD']: for path in ['/health', '/v1/me', '/v1/sync', '/app.js', '/v1/auth/login']: record((method, path), server.request(method, path), ('allow', 'content-type')) for path in ['/index.html', '/a/../app.js', '//app.js', '/app.js/', '/v1', '/v1/', '/v1//me', '/v1/../app.js', '/%61pp.js', '/%2e%2e/app.js']: record(('path', path), server.request('GET', path), ('location', 'content-type', 'cache-control')) for range_value in ['bytes=0-9', 'bytes=-10', 'bytes=10-', 'bytes=99999999-', 'bytes=bad', 'items=0-1', 'bytes=10-1', 'bytes=0-0,2-2']: result = server.request('GET', '/icon.svg', headers={'Range': range_value}) if result[0] == 206 and 'multipart/' in {k.lower():v for k,v in result[1]}.get('content-type', ''): # Multipart boundaries are randomly generated in the reference. assert b'Content-Range: bytes 0-0/' in result[2] assert b'Content-Range: bytes 2-2/' in result[2] observed[('range', range_value)] = result[0] else: record(('range', range_value), result, ('content-range', 'accept-ranges', 'content-type')) raw_sync = [b'null', b'{} trailing', b'{"CURSOR":null,"CHANGES":null}', b'{"cursor":0,"cursor":null}', b'{"cursor":1.0}', b'[]', b'{"changes":[null]}', b'{"deviceId":false}', b'{"pull":0}', b'{"changes":[],"changes":null}', b'{"cursor":9223372036854775808}'] for n, raw in enumerate(raw_sync): record(('raw sync', n), server.request('POST', '/v1/sync', token=token, raw=raw)) cases = [change('x', entity='Category'), change('x'*129), change('x', payload=None), change('x', payload=[1,2,3]), change('x', stamp='0001-01-01T00:00:00Z'), change('x', stamp='2025-01-01t00:00:00z'), change('x', stamp='2025-01-01T0:00:00Z'), change('x', stamp='2025-01-01T00:00:60Z'), change('x', stamp='2025-01-01T00:00:00,123456789123Z'), change('x', stamp='2025-01-01T00:00:00+24:00'), change('x', stamp='2025-01-01T00:00:00+00:60'), change('x', stamp='2025-01-01T00:00:00.123456789123Z'), change('x', stamp='9999-01-01T00:00:00Z'), change('x', payload='x'*(256*1024))] for n, value in enumerate(cases): record(('sync validation', n), server.request('POST', '/v1/sync', {'changes':[value]}, token)) for n, raw in enumerate([b'null', b'{} tail', b'{"refreshToken":null}', b'{"REFRESHTOKEN":"bad"}', b'{"refreshToken":1}', b'{"extra":1}']): record(('refresh JSON', n), server.request('POST', '/v1/auth/refresh', raw=raw)) record(('logout JSON', n), server.request('POST', '/v1/auth/logout', raw=raw)) for auth in ['', 'Basic bad', 'bearer bad', 'Bearer ', 'Bearer bad']: record(('bearer', auth), server.request('GET', '/v1/me', headers={'Authorization':auth})) # Verify signed JWTs with claims from Go's permissive JSON decoder. for n, claims in enumerate([ {'sub':'u','email':'e','iss':'balance-server','exp':4102444800}, {'SUB':'u','EMAIL':'e','ISS':'balance-server','EXP':4102444800}, {'sub':'u','email':None,'iss':'balance-server','exp':4102444800}, {'sub':'u','iss':'balance-server','exp':4102444800,'unknown':123}, {'sub':'u','iss':'wrong','exp':4102444800}, {'sub':'u','iss':'balance-server','exp':1}, {'sub':'','iss':'balance-server','exp':4102444800}, ]): encode = lambda data: base64.urlsafe_b64encode(data).rstrip(b'=') unsigned = b'ignored-header.' + encode(json.dumps(claims).encode()) signed = (unsigned+b'.'+encode(hmac.new(SECRET.encode(), unsigned, hashlib.sha256).digest())).decode() record(('signed claims', n), server.request('GET', '/v1/me', token=signed)) # Only one simultaneous rotation of a given refresh session may succeed. fresh = expect(server.request('POST','/v1/auth/login',credentials()),200) with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: statuses = sorted(pool.map(lambda _: server.request('POST','/v1/auth/refresh',{'refreshToken':fresh['refreshToken']})[0], range(4))) assert statuses == [200,401,401,401], statuses with running(binary, database) as server: for n, raw in enumerate([b'null', b'{"email":null,"password":null}', b'{"EMAIL":"bad","PASSWORD":"abcdefgh"}', b'{"email":"bad","email":null}', b'{"email":false}', b'{"email":"bad"} trailing', b'{"email":"bad","unknown":1}']): for endpoint in ['register','login']: record((endpoint, n), server.request('POST', '/v1/auth/'+endpoint, raw=raw)) for n, value in enumerate([credentials('a@b'),dict(email='utf8@example.com',password='я'*4), dict(email='long@example.com',password='x'*1025)]): result = server.request('POST','/v1/auth/register',value) if result[0] == 201: assert expect(server.request('POST','/v1/auth/login',value),200)['user'] == result[2]['user'] observed[('password boundary',n)] = result[0] else: record(('password boundary',n), result) with running(binary,database,BALANCE_REFRESH_TTL='1ms') as server: session = expect(server.request('POST','/v1/auth/login',credentials()),200) time.sleep(.05) record('expired refresh',server.request('POST','/v1/auth/refresh',{'refreshToken':session['refreshToken']})) return observed def raw_json_scenarios(binary, database): """RawMessage must retain numbers, duplicate keys and unusual Unicode bytes.""" observed = {} with running(binary, database) as server: session = expect(server.request('POST','/v1/auth/register',credentials()),201) token = session['accessToken'] cursor = 0 for n, payload in enumerate([ b'{ "duplicate":1, "duplicate":2, "n":123456789012345678901234567890, "big":1e999 }', br'{"lone":"\ud800","low":"\udfff","pair":"\ud83d\ude00"}', b'{"invalid":"\xff\xc0\xed\xa0\x80"}', b'{"html":"<>&","separator":"\xe2\x80\xa8\xe2\x80\xa9"}', br'{"nested":[{"payload":{"id":"inner"}},"quote\"end",null,true]}', ]): raw = (b'{"cursor":'+str(cursor).encode()+b',"changes":[{"entity":"transaction","id":"wire-'+str(n).encode()+ b'","updatedAt":"2025-01-01T00:00:00Z","payload":'+payload+b'}]}') result = server.request('POST','/v1/sync',token=token,raw=raw,decode_json=False) assert result[0] == 200, result data = result[2] # Latin-1 preserves byte indices while Python locates the complete JSON value. start = data.index(b'"payload":') + len(b'"payload":') _, end = json.JSONDecoder().raw_decode(data[start:].decode('latin-1')) observed[('wire payload',n)] = data[start:start+end] decoded = json.loads(data.decode('utf-8','replace')) cursor = decoded['cursor'] assert len(decoded['changes']) == 1 for n, raw in enumerate([br'{"email":"\ud800","password":"abcdefgh"}', b'{"email":"\xff@example.com","password":"abcdefgh"}']): # Login has deterministic invalid credentials and exercises scalar normalization. result = server.request('POST','/v1/auth/login',raw=raw) observed[('unicode credentials',n)] = (result[0],result[2]) for n, raw_id in enumerate([br'"\ud800"',b'"\xff"']): raw = (b'{"cursor":'+str(cursor).encode()+b',"changes":[{"entity":"category","id":'+raw_id+ b',"updatedAt":"2025-01-02T00:00:00Z","payload":{}}]}') result = server.request('POST','/v1/sync',token=token,raw=raw) assert result[0] == 200, result data = result[2] data.pop('serverTime') observed[('unicode id',n)] = data cursor = data['cursor'] with sqlite3.connect(database) as db: db.text_factory = bytes observed['raw database payloads'] = db.execute('SELECT record_id,payload FROM records ORDER BY record_id').fetchall() return observed def web_assets(value): """Stand in for embedded web files, so the reference survives web client edits.""" if isinstance(value, bytes): for asset in sorted((ROOT / 'web').iterdir()): if value == asset.read_bytes(): return ('web asset', asset.name) return value if isinstance(value, dict): return {k: web_assets(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return type(value)(web_assets(v) for v in value) return value def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--binary', type=Path, default=ROOT / 'target/debug/balance-server') parser.add_argument('--update', action='store_true', help=f'record the responses as the new reference in {EXPECTED.name}') args = parser.parse_args() binary = args.binary.resolve() assert binary.is_file(), f'Build first: cargo build --locked (missing {binary})' with tempfile.TemporaryDirectory(prefix='balance-compatibility-') as directory: directory = Path(directory) actual = {} print('Running HTTP/JSON scenarios...', flush=True) actual['scenario'] = scenario(binary, directory / 'scenario.db') print('Running HTTP/JSON edge cases...', flush=True) actual['edge'] = edge_scenarios(binary, directory / 'edge.db') print('Running raw JSON payload and Unicode preservation...', flush=True) actual['raw'] = raw_json_scenarios(binary, directory / 'raw.db') actual = web_assets(actual) if args.update: EXPECTED.write_text(pprint.pformat(actual, width=100) + '\n') print(f'Recorded {EXPECTED.relative_to(ROOT)}') else: expected = ast.literal_eval(EXPECTED.read_text()) for group in expected: for key in expected[group]: assert actual[group].get(key) == expected[group][key], ( f'Reference mismatch: {group} {key}\n' f'Expected: {expected[group][key]!r}\nActual: {actual[group].get(key)!r}') assert actual[group].keys() == expected[group].keys(), f'Scenario set changed in {group}; re-record with --update' print('Checking SQLite persistence across restarts...', flush=True) takeover(binary, binary, directory / 'restart.db') print('PASS: HTTP/JSON reference scenarios and SQLite persistence') if __name__ == '__main__': main()