+108

convey/auth.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """Shared pairing auth helpers.""" 5 5 + 6 6 + from __future__ import annotations 7 7 + 8 8 + import logging 9 9 + from functools import wraps 10 10 + from typing import Any, Callable, TypeVar, cast 11 11 + 12 12 + from flask import g, jsonify, request, session 13 13 + from werkzeug.security import check_password_hash 14 14 + 15 15 + from think.pairing.devices import Device, find_device_by_session_key_hash 16 16 + from think.pairing.keys import hash_session_key, mask_session_key 17 17 + from think.utils import get_config 18 18 + 19 19 + logger = logging.getLogger(__name__) 20 20 + 21 21 + F = TypeVar("F", bound=Callable[..., Any]) 22 22 + 23 23 + 24 24 + def extract_bearer_token() -> str | None: 25 25 + header = request.headers.get("Authorization") 26 26 + if not isinstance(header, str): 27 27 + return None 28 28 + scheme, _, value = header.partition(" ") 29 29 + if scheme.lower() != "bearer": 30 30 + return None 31 31 + token = value.strip() 32 32 + return token or None 33 33 + 34 34 + 35 35 + def resolve_paired_device() -> Device | None: 36 36 + token = extract_bearer_token() 37 37 + if token is None: 38 38 + return None 39 39 + masked = mask_session_key(token) 40 40 + session_key_hash = hash_session_key(token) 41 41 + device = find_device_by_session_key_hash(session_key_hash) 42 42 + if device is None: 43 43 + logger.debug("paired device not found session_key=%s", masked) 44 44 + return None 45 45 + logger.debug("paired device resolved id=%s session_key=%s", device["id"], masked) 46 46 + return device 47 47 + 48 48 + 49 49 + def _check_basic_auth() -> bool: 50 50 + auth = request.authorization 51 51 + if not auth or auth.type != "basic": 52 52 + return False 53 53 + password_hash = str(get_config().get("convey", {}).get("password_hash", "") or "") 54 54 + if not password_hash: 55 55 + return False 56 56 + return check_password_hash(password_hash, auth.password or "") 57 57 + 58 58 + 59 59 + def _is_setup_complete() -> bool: 60 60 + return bool(get_config().get("setup", {}).get("completed_at")) 61 61 + 62 62 + 63 63 + def is_owner_authed() -> bool: 64 64 + if session.get("logged_in"): 65 65 + return True 66 66 + if _check_basic_auth(): 67 67 + return True 68 68 + if not _is_setup_complete(): 69 69 + return False 70 70 + config = get_config() 71 71 + if not config.get("convey", {}).get("trust_localhost", False): 72 72 + return False 73 73 + remote_addr = request.remote_addr 74 74 + is_localhost = remote_addr in ("127.0.0.1", "::1", "localhost") 75 75 + proxy_headers = ( 76 76 + request.headers.get("X-Forwarded-For") 77 77 + or request.headers.get("X-Real-IP") 78 78 + or request.headers.get("X-Forwarded-Host") 79 79 + ) 80 80 + return bool(is_localhost and not proxy_headers) 81 81 + 82 82 + 83 83 + def require_paired_device(func: F) -> F: 84 84 + @wraps(func) 85 85 + def wrapped(*args: Any, **kwargs: Any) -> Any: 86 86 + device = resolve_paired_device() 87 87 + if device is None: 88 88 + return ( 89 89 + jsonify( 90 90 + { 91 91 + "error": "paired device required", 92 92 + "reason": "auth_required", 93 93 + } 94 94 + ), 95 95 + 401, 96 96 + ) 97 97 + g.paired_device = device 98 98 + return func(*args, **kwargs) 99 99 + 100 100 + return cast(F, wrapped) 101 101 + 102 102 + 103 103 + __all__ = [ 104 104 + "extract_bearer_token", 105 105 + "is_owner_authed", 106 106 + "require_paired_device", 107 107 + "resolve_paired_device", 108 108 + ]

+174

tests/test_pairing_auth.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + from __future__ import annotations 5 5 + 6 6 + import base64 7 7 + import json 8 8 + 9 9 + from flask import Flask, g, jsonify, session 10 10 + from werkzeug.security import generate_password_hash 11 11 + 12 12 + from convey.auth import ( 13 13 + extract_bearer_token, 14 14 + is_owner_authed, 15 15 + require_paired_device, 16 16 + resolve_paired_device, 17 17 + ) 18 18 + from think.pairing.devices import register_device 19 19 + from think.pairing.keys import hash_session_key 20 20 + 21 21 + 22 22 + def _write_config(journal_copy, payload: dict) -> None: 23 23 + (journal_copy / "config" / "journal.json").write_text( 24 24 + json.dumps(payload), encoding="utf-8" 25 25 + ) 26 26 + 27 27 + 28 28 + def _read_config(journal_copy) -> dict: 29 29 + return json.loads((journal_copy / "config" / "journal.json").read_text("utf-8")) 30 30 + 31 31 + 32 32 + def _create_app() -> Flask: 33 33 + app = Flask(__name__) 34 34 + app.secret_key = "test-secret" 35 35 + app.config["TESTING"] = True 36 36 + 37 37 + @app.get("/bearer") 38 38 + def bearer_view(): 39 39 + return jsonify( 40 40 + { 41 41 + "token": extract_bearer_token(), 42 42 + "device_id": (resolve_paired_device() or {}).get("id"), 43 43 + } 44 44 + ) 45 45 + 46 46 + @app.get("/owner") 47 47 + def owner_view(): 48 48 + return jsonify({"owner_authed": is_owner_authed()}) 49 49 + 50 50 + @app.get("/paired") 51 51 + @require_paired_device 52 52 + def paired_view(): 53 53 + return jsonify({"device_id": g.paired_device["id"]}) 54 54 + 55 55 + @app.get("/session-login") 56 56 + def session_login(): 57 57 + session["logged_in"] = True 58 58 + session.permanent = True 59 59 + return jsonify({"ok": True}) 60 60 + 61 61 + return app 62 62 + 63 63 + 64 64 + def test_extract_bearer_token_handles_missing_and_malformed_headers(journal_copy): 65 65 + app = _create_app() 66 66 + client = app.test_client() 67 67 + 68 68 + assert client.get("/bearer").get_json() == {"token": None, "device_id": None} 69 69 + assert client.get("/bearer", headers={"Authorization": "Bearer"}).get_json() == { 70 70 + "token": None, 71 71 + "device_id": None, 72 72 + } 73 73 + assert client.get("/bearer", headers={"Authorization": "Basic abc"}).get_json() == { 74 74 + "token": None, 75 75 + "device_id": None, 76 76 + } 77 77 + 78 78 + 79 79 + def test_resolve_paired_device_requires_matching_hash(journal_copy): 80 80 + app = _create_app() 81 81 + client = app.test_client() 82 82 + register_device( 83 83 + name="Phone", 84 84 + platform="ios", 85 85 + public_key="ssh-ed25519 AAAAauth", 86 86 + session_key_hash=hash_session_key("dsk_real"), 87 87 + bundle_id="org.solpbc.solstone-swift", 88 88 + app_version="0.1.0", 89 89 + paired_at="2026-04-20T15:31:02Z", 90 90 + ) 91 91 + 92 92 + missing = client.get("/bearer", headers={"Authorization": "Bearer dsk_wrong"}) 93 93 + found = client.get("/bearer", headers={"Authorization": "Bearer dsk_real"}) 94 94 + 95 95 + assert missing.get_json() == {"token": "dsk_wrong", "device_id": None} 96 96 + assert found.get_json()["token"] == "dsk_real" 97 97 + assert found.get_json()["device_id"].startswith("dev_") 98 98 + 99 99 + 100 100 + def test_is_owner_authed_via_basic_auth(journal_copy): 101 101 + payload = _read_config(journal_copy) 102 102 + payload["convey"]["password_hash"] = generate_password_hash("test123") 103 103 + payload["setup"] = {"completed_at": 1700000000000} 104 104 + _write_config(journal_copy, payload) 105 105 + 106 106 + app = _create_app() 107 107 + client = app.test_client() 108 108 + creds = base64.b64encode(b":test123").decode("ascii") 109 109 + 110 110 + response = client.get( 111 111 + "/owner", 112 112 + headers={ 113 113 + "Authorization": f"Basic {creds}", 114 114 + "X-Forwarded-For": "1.2.3.4", 115 115 + }, 116 116 + ) 117 117 + 118 118 + assert response.get_json() == {"owner_authed": True} 119 119 + 120 120 + 121 121 + def test_is_owner_authed_via_session_cookie(journal_copy): 122 122 + app = _create_app() 123 123 + client = app.test_client() 124 124 + 125 125 + client.get("/session-login") 126 126 + response = client.get("/owner") 127 127 + 128 128 + assert response.get_json() == {"owner_authed": True} 129 129 + 130 130 + 131 131 + def test_is_owner_authed_via_trust_localhost(journal_copy): 132 132 + payload = _read_config(journal_copy) 133 133 + payload["convey"]["trust_localhost"] = True 134 134 + payload["setup"] = {"completed_at": 1700000000000} 135 135 + payload["convey"].pop("password_hash", None) 136 136 + _write_config(journal_copy, payload) 137 137 + 138 138 + app = _create_app() 139 139 + client = app.test_client() 140 140 + response = client.get("/owner") 141 141 + 142 142 + assert response.get_json() == {"owner_authed": True} 143 143 + 144 144 + 145 145 + def test_require_paired_device_returns_401_json_without_bearer(journal_copy): 146 146 + app = _create_app() 147 147 + client = app.test_client() 148 148 + 149 149 + response = client.get("/paired") 150 150 + 151 151 + assert response.status_code == 401 152 152 + assert response.get_json() == { 153 153 + "error": "paired device required", 154 154 + "reason": "auth_required", 155 155 + } 156 156 + 157 157 + 158 158 + def test_require_paired_device_sets_g_paired_device(journal_copy): 159 159 + device = register_device( 160 160 + name="Phone", 161 161 + platform="ios", 162 162 + public_key="ssh-ed25519 AAAApaired", 163 163 + session_key_hash=hash_session_key("dsk_paired"), 164 164 + bundle_id="org.solpbc.solstone-swift", 165 165 + app_version="0.1.0", 166 166 + paired_at="2026-04-20T15:31:02Z", 167 167 + ) 168 168 + app = _create_app() 169 169 + client = app.test_client() 170 170 + 171 171 + response = client.get("/paired", headers={"Authorization": "Bearer dsk_paired"}) 172 172 + 173 173 + assert response.status_code == 200 174 174 + assert response.get_json() == {"device_id": device["id"]}