226 lines
13 KiB
Python
226 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Compare two isolated 2FAuth instances without emitting response bodies/secrets.
|
|
No redirects or retries. Cookie sessions require explicit opt-in. See compatibility/README.md.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import copy
|
|
import http.cookiejar
|
|
import re
|
|
import os
|
|
from pathlib import Path
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
DEFAULT_CASES = [
|
|
{'id': 'health', 'method': 'GET', 'path': '/up'},
|
|
{'id': 'manifest', 'method': 'GET', 'path': '/manifest.json'},
|
|
{'id': 'deprecated-name', 'method': 'GET', 'path': '/api/v1/user/name'},
|
|
{'id': 'unknown-api', 'method': 'GET', 'path': '/api/v1/__compat_missing__'},
|
|
{'id': 'cors-preflight', 'method': 'OPTIONS', 'path': '/api/v1/twofaccounts', 'expected_status': 204,
|
|
'headers': {'Origin': 'https://example.invalid', 'Access-Control-Request-Method': 'POST', 'Access-Control-Request-Headers': 'Authorization,Content-Type'},
|
|
'compare_headers': ['Access-Control-Allow-Origin', 'Access-Control-Allow-Methods', 'Access-Control-Allow-Headers', 'Access-Control-Allow-Credentials'],
|
|
'expected_headers': {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': None}},
|
|
{'id': 'cors-actual', 'method': 'GET', 'path': '/api/v1/twofaccounts', 'expected_status': 401,
|
|
'headers': {'Origin': 'https://example.invalid'},
|
|
'compare_headers': ['Access-Control-Allow-Origin', 'Access-Control-Allow-Credentials'],
|
|
'expected_headers': {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': None}},
|
|
{'id': 'cors-without-origin', 'method': 'GET', 'path': '/api/v1/twofaccounts', 'expected_status': 401,
|
|
'compare_headers': ['Access-Control-Allow-Origin', 'Access-Control-Allow-Credentials'],
|
|
'expected_headers': {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': None}},
|
|
]
|
|
SAFE_CASES = {(c.get('method', 'GET'), c['path']) for c in DEFAULT_CASES}
|
|
MAX_BYTES = 4 * 1024 * 1024
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, *args, **kwargs): return None
|
|
|
|
def base_url(value):
|
|
parsed = urllib.parse.urlsplit(value)
|
|
if parsed.scheme not in ('http', 'https') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
|
|
raise ValueError('base must be HTTP(S), without credentials, query or fragment')
|
|
return value.rstrip('/')
|
|
|
|
def fetch(base, case, side, timeout, opener=None, cookies=None):
|
|
headers = {'Accept': 'application/json', 'X-Requested-With': 'XMLHttpRequest'}
|
|
headers.update(case.get('headers', {}))
|
|
# Values never appear in output. Side-specific credentials refer to env names.
|
|
env_headers = case.get('headers_env', {}).get(side, {})
|
|
for header, name in env_headers.items():
|
|
if name not in os.environ: return {'error': 'missing_environment_variable'}
|
|
headers[header] = os.environ[name]
|
|
for header, cookie_name in case.get('headers_from_cookies', {}).items():
|
|
matches = [c for c in (cookies or []) if c.name == cookie_name]
|
|
if len(matches) != 1: return {'error': 'missing_or_ambiguous_cookie'}
|
|
headers[header] = urllib.parse.unquote(matches[0].value)
|
|
data = None
|
|
if sum(name in case for name in ('json','form','raw')) > 1:
|
|
raise ValueError('a request must choose one body encoding')
|
|
if 'json' in case:
|
|
data = json.dumps(case['json']).encode()
|
|
headers['Content-Type'] = 'application/json'
|
|
elif 'form' in case:
|
|
data = urllib.parse.urlencode(case['form'], doseq=True).encode()
|
|
headers['Content-Type'] = 'application/x-www-form-urlencoded'
|
|
elif 'raw' in case:
|
|
data = case['raw'].encode()
|
|
req = urllib.request.Request(base + case['path'], data=data, headers=headers, method=case.get('method', 'GET'))
|
|
try:
|
|
try: response = (opener or urllib.request.build_opener(NoRedirect())).open(req, timeout=timeout)
|
|
except urllib.error.HTTPError as exc: response = exc
|
|
with response:
|
|
body = response.read(MAX_BYTES + 1)
|
|
if len(body) > MAX_BYTES: return {'error': 'response_too_large'}
|
|
result = {'status': response.code, 'content_type': response.headers.get('Content-Type', '').split(';')[0].lower(), 'body': body, 'headers': response.headers}
|
|
return result
|
|
except Exception as exc:
|
|
# Exception text can contain URLs or credentials, so only emit class name.
|
|
return {'error': type(exc).__name__}
|
|
|
|
def remove_pointer(value, pointer):
|
|
if not pointer.startswith('/'): raise ValueError('ignored JSON pointers must start with /')
|
|
parts = [s.replace('~1', '/').replace('~0', '~') for s in pointer[1:].split('/')]
|
|
cursor = value
|
|
for part in parts[:-1]:
|
|
try: cursor = cursor[int(part)] if isinstance(cursor, list) else cursor[part]
|
|
except (KeyError, IndexError, ValueError, TypeError): return
|
|
if isinstance(cursor, dict): cursor.pop(parts[-1], None)
|
|
elif isinstance(cursor, list):
|
|
try: cursor[int(parts[-1])] = None
|
|
except (IndexError, ValueError): pass
|
|
|
|
def comparable(response, base, case):
|
|
body = response['body'].decode('utf-8', errors='replace')
|
|
if response['content_type'] == 'application/json':
|
|
try:
|
|
body = json.loads(body)
|
|
def normalize(value):
|
|
if isinstance(value, str): return value.replace(base, '<BASE>')
|
|
if isinstance(value, dict): return {key: normalize(item) for key, item in value.items()}
|
|
if isinstance(value, list): return [normalize(item) for item in value]
|
|
return value
|
|
body = normalize(body)
|
|
for pointer in case.get('ignore_json_pointers', []): remove_pointer(body, pointer)
|
|
except json.JSONDecodeError: body = body.replace(base, '<BASE>')
|
|
else:
|
|
body = body.replace(base, '<BASE>')
|
|
return body
|
|
|
|
|
|
def resolve(value, variables):
|
|
"""Resolve explicit {{capture}} references, keeping typed whole-value JSON."""
|
|
if isinstance(value, dict): return {k: resolve(v, variables) for k, v in value.items()}
|
|
if isinstance(value, list): return [resolve(v, variables) for v in value]
|
|
if not isinstance(value, str): return value
|
|
match = re.fullmatch(r'\{\{([A-Za-z_][A-Za-z0-9_]*)\}\}', value)
|
|
if match: return variables[match[1]]
|
|
return re.sub(r'\{\{([A-Za-z_][A-Za-z0-9_]*)\}\}', lambda m: str(variables[m[1]]), value)
|
|
|
|
|
|
def read_pointer(value, pointer):
|
|
if pointer and not pointer.startswith('/'): raise ValueError('invalid JSON pointer')
|
|
for token in pointer[1:].split('/') if pointer else []:
|
|
token = token.replace('~1', '/').replace('~0', '~')
|
|
value = value[int(token)] if isinstance(value, list) else value[token]
|
|
return value
|
|
|
|
|
|
def is_text_token(response):
|
|
return response.get('status') == 200 and response.get('content_type') in ('text/plain', 'text/html') and re.fullmatch(rb'[A-Za-z0-9_-]{32,256}', response.get('body', b'')) is not None
|
|
|
|
|
|
def assert_json_response(response, case):
|
|
try:
|
|
payload = json.loads(response['body'])
|
|
for pointer, expected in case.get('expected_json', {}).items():
|
|
if read_pointer(payload, pointer) != expected: return False
|
|
for pointer in case.get('absent_json_pointers', []):
|
|
try: read_pointer(payload, pointer)
|
|
except (KeyError, IndexError, TypeError, ValueError): continue
|
|
return False
|
|
return True
|
|
except (KeyError, IndexError, TypeError, ValueError):
|
|
return False
|
|
|
|
|
|
def capture_values(response, captures, variables):
|
|
if not captures: return
|
|
payload = json.loads(response['body'])
|
|
for name, pointer in captures.items():
|
|
if not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*', name): raise ValueError('invalid capture name')
|
|
if pointer and not pointer.startswith('/'): raise ValueError('invalid capture pointer')
|
|
value = payload
|
|
for token in pointer[1:].split('/') if pointer else []:
|
|
token = token.replace('~1', '/').replace('~0', '~')
|
|
value = value[int(token)] if isinstance(value, list) else value[token]
|
|
variables[name] = value
|
|
|
|
|
|
def validate_path(path):
|
|
if not isinstance(path, str) or not path.startswith('/') or path.startswith('//') or any(c in path for c in '#\r\n'):
|
|
raise ValueError('case paths must be absolute local paths without fragments or newlines')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--reference', required=True, type=base_url)
|
|
parser.add_argument('--candidate', required=True, type=base_url)
|
|
parser.add_argument('--cases', type=Path)
|
|
parser.add_argument('--allow-side-effects', action='store_true')
|
|
parser.add_argument('--timeout', type=float, default=10)
|
|
parser.add_argument('--sessions', action='store_true', help='retain separate in-memory cookie jars; requires --allow-side-effects')
|
|
args = parser.parse_args()
|
|
cases = json.loads(args.cases.read_text()) if args.cases else DEFAULT_CASES
|
|
if args.reference == args.candidate: parser.error('reference and candidate must be separate instances')
|
|
if args.sessions and not args.allow_side_effects: parser.error('--sessions requires --allow-side-effects')
|
|
cookies = {side: http.cookiejar.CookieJar() for side in ('reference', 'candidate')}
|
|
openers = {side: urllib.request.build_opener(NoRedirect(), urllib.request.HTTPCookieProcessor(cookies[side])) for side in cookies}
|
|
variables = {side: {} for side in cookies}
|
|
failed = False
|
|
for index, case in enumerate(cases):
|
|
path = case.get('path', '')
|
|
try: validate_path(path)
|
|
except ValueError as error: parser.error(str(error))
|
|
risky = (case.get('method', 'GET'), path) not in SAFE_CASES or bool(case.get('headers_env')) or bool(case.get('headers_from_cookies')) or args.sessions or (bool(case.get('headers')) and case not in DEFAULT_CASES)
|
|
# User case labels can contain credentials too. Output stable numeric IDs only.
|
|
if (risky or case.get('side_effects', False)) and not args.allow_side_effects:
|
|
print(json.dumps({'case': index, 'result': 'skipped_requires_allow_side_effects'})); continue
|
|
responses = []
|
|
for side, base in (('reference', args.reference), ('candidate', args.candidate)):
|
|
try:
|
|
resolved = resolve(copy.deepcopy(case), variables[side])
|
|
validate_path(resolved['path'])
|
|
response = fetch(base, resolved, side, args.timeout,
|
|
openers[side] if args.sessions else None,
|
|
cookies[side] if args.sessions else None)
|
|
if 'error' not in response:
|
|
capture_values(response, case.get('capture_json', {}), variables[side])
|
|
except (KeyError, ValueError, TypeError, IndexError):
|
|
response = {'error': 'invalid_or_missing_capture'}
|
|
responses.append(response)
|
|
left, right = responses
|
|
errors = [r.get('error') for r in (left, right)]
|
|
if any(errors):
|
|
failed = True
|
|
print(json.dumps({'case': index, 'result': 'error', 'errors': errors})); continue
|
|
checks = {
|
|
'status': left['status'] == right['status'],
|
|
'content_type': left['content_type'] == right['content_type'],
|
|
'body': comparable(left, args.reference, case) == comparable(right, args.candidate, case),
|
|
}
|
|
if case.get('expect_text_token'):
|
|
checks['text_token_shape'] = all(is_text_token(response) for response in (left, right))
|
|
if case.get('expected_json') or case.get('absent_json_pointers'):
|
|
checks['json_assertions'] = all(assert_json_response(response, case) for response in (left, right))
|
|
if 'expected_status' in case:
|
|
checks['expected_status'] = left['status'] == right['status'] == case['expected_status']
|
|
if 'expected_headers' in case:
|
|
checks['expected_headers'] = all(response['headers'].get(header) == expected for response in (left, right) for header, expected in case['expected_headers'].items())
|
|
for header in case.get('compare_headers', []):
|
|
# Headers are compared in memory; neither names nor values are printed.
|
|
checks['selected_headers'] = checks.get('selected_headers', True) and left['headers'].get(header) == right['headers'].get(header)
|
|
equal = all(checks.values()); failed |= not equal
|
|
print(json.dumps({'case': index, 'result': 'equal' if equal else 'different', 'reference_status': left['status'], 'candidate_status': right['status'], 'checks': checks}))
|
|
return int(failed)
|
|
if __name__ == '__main__': raise SystemExit(main())
|