From 34fbe0d65293e28def9c314ce8a47ef3c3b7670a Mon Sep 17 00:00:00 2001 From: Max Ustinov Date: Fri, 16 Oct 2026 10:33:14 -0700 Subject: [PATCH] feat(rest_api): fetch a self-managed server's public key from /api/rest/server_public_key --- keepercommander/rest_api.py | 74 +++++++++++-- unit-tests/test_server_public_key_fetch.py | 114 +++++++++++++++++++++ 2 files changed, 177 insertions(+), 11 deletions(-) create mode 100644 unit-tests/test_server_public_key_fetch.py diff --git a/keepercommander/rest_api.py b/keepercommander/rest_api.py index 56be0c9ec..9b5d4f872 100644 --- a/keepercommander/rest_api.py +++ b/keepercommander/rest_api.py @@ -18,7 +18,7 @@ import time import sys -from typing import Union, Dict, Optional +from typing import Union, Dict, Optional, Set from .params import RestApiContext from .error import KeeperApiError, Error @@ -121,6 +121,20 @@ } # type: Dict[int, Union[rsa.RSAPublicKey, ec.EllipticCurvePublicKey, Dict[str, Union[ec.EllipticCurvePublicKey, bytes]]]] +def _register_server_public_key(key_id, key_value): + # type: (int, str) -> None + """Use key_value (base64url uncompressed EC point) for key_id, in Commander and in the KSM SDK.""" + SERVER_PUBLIC_KEYS[key_id] = crypto.load_ec_public_key(utils.base64_url_decode(key_value)) + # The embedded KSM SDK (ksm app client add, slack-app-setup, service-mode docker + # setup) keeps its own copy of the key table. Without this it answers the server's + # key_id request with Keeper's key and gets "encrypted transmission key is invalid". + try: + from keeper_secrets_manager_core.keeper_globals import keeper_public_keys + keeper_public_keys[str(key_id)] = key_value + except ImportError: + pass + + def _load_custom_server_public_key(): # Self-hosted/on-premise Keeper deployments (e.g. Self-Managed Keeper) generate # their own transmission-key keypair at install time; it is never one of @@ -139,18 +153,47 @@ def _load_custom_server_public_key(): logging.warning(f"KEEPER_SERVER_PUBLIC_KEY_ID='{key_id_str}' is not a valid integer; ignoring custom server public key.") return try: - SERVER_PUBLIC_KEYS[key_id] = crypto.load_ec_public_key(utils.base64_url_decode(key_value)) + _register_server_public_key(key_id, key_value) except Exception as e: logging.warning(f"Failed to load KEEPER_SERVER_PUBLIC_KEY as an EC public key ({e}); ignoring custom server public key.") - return - # The embedded KSM SDK (ksm app client add, slack-app-setup, service-mode docker - # setup) keeps its own copy of the key table. Without this it answers the server's - # key_id request with Keeper's key and gets "encrypted transmission key is invalid". + + +# Servers already asked for their key, so each is asked at most once per process. +_server_key_fetch_attempted = set() # type: Set[str] + + +def _fetch_server_public_key(context): + # type: (RestApiContext) -> Optional[int] + """Learn a self-managed server's transmission key from GET /api/rest/server_public_key. + + Self-Managed Keeper serves {"serverPublicKeyId", "serverPublicKey", "fingerprint"} there. + Keeper's regions do not have the endpoint; this is only called after a server has + rejected the built-in keys, and any failure returns None so the caller keeps its + existing behaviour. TLS verification (context.certificate_check) is what makes the + answer trustworthy, exactly as for every other request to the same server. + """ + url = context.server_base + 'server_public_key' + if url in _server_key_fetch_attempted: + return None + _server_key_fetch_attempted.add(url) try: - from keeper_secrets_manager_core.keeper_globals import keeper_public_keys - keeper_public_keys[str(key_id)] = key_value - except ImportError: - pass + rs = requests.get(url, proxies=context.proxies, verify=context.certificate_check, timeout=(5, 10)) + if rs.status_code != 200: + logging.debug('Server public key endpoint %s answered HTTP %d', url, rs.status_code) + return None + body = rs.json() + key_id = int(body['serverPublicKeyId']) + key_value = body['serverPublicKey'] + if not (7 <= key_id <= 18) or not isinstance(key_value, str): + logging.debug('Server public key endpoint %s returned an unusable key id %s', url, key_id) + return None + _register_server_public_key(key_id, key_value) + except Exception as e: + logging.debug('Could not fetch server public key from %s: %s', url, e) + return None + logging.info('Using the server public key published by %s (key id %d, %s)', + context.server_base, key_id, body.get('fingerprint') or 'no fingerprint') + return key_id _load_custom_server_public_key() @@ -286,6 +329,14 @@ def execute_rest(context, endpoint, payload, timeout=None): elif server_key_id != context.server_key_id: # If server returns non-EC key without qrc_ec_key_id, disable QRC if not (7 <= server_key_id <= 18): + # A self-managed server rejects Keeper's built-in keys. Ask it for its + # own key once; servers without the endpoint fall through unchanged. + fetched_key_id = _fetch_server_public_key(context) + if fetched_key_id is not None: + context.disable_qrc() + context.server_key_id = fetched_key_id + run_request = True + continue key_mismatch_retries += 1 if key_mismatch_retries > max_key_mismatch_retries: raise KeeperApiError( @@ -294,7 +345,8 @@ def execute_rest(context, endpoint, payload, timeout=None): "This usually means you're connecting to a self-hosted/on-premise Keeper server that " "uses its own transmission-key keypair, not one of Commander's built-in keys. Set the " "KEEPER_SERVER_PUBLIC_KEY_ID and KEEPER_SERVER_PUBLIC_KEY environment variables to that " - "deployment's key id/public key.") + "deployment's key id/public key, or use a server that publishes it at " + "/api/rest/server_public_key.") logging.warning(f"Server returned non-EC key_id={server_key_id} without qrc_ec_key_id. Falling back to EC-only encryption.") context.disable_qrc() else: diff --git a/unit-tests/test_server_public_key_fetch.py b/unit-tests/test_server_public_key_fetch.py new file mode 100644 index 000000000..6df55b652 --- /dev/null +++ b/unit-tests/test_server_public_key_fetch.py @@ -0,0 +1,114 @@ +import json +import unittest +from unittest import mock + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec + +from keepercommander import crypto, rest_api, utils +from keepercommander.params import RestApiContext +from keepercommander.proto import APIRequest_pb2 as proto + + +def _new_server_key(): + private_key = ec.generate_private_key(ec.SECP256R1()) + point = private_key.public_key().public_bytes(serialization.Encoding.X962, + serialization.PublicFormat.UncompressedPoint) + return private_key, utils.base64_url_encode(point) + + +def _response(status, body=None, content_type='application/json', content=b''): + rs = mock.Mock() + rs.status_code = status + rs.headers = {'Content-Type': content_type} + rs.json.return_value = body + rs.content = content + return rs + + +class TestServerPublicKeyFetch(unittest.TestCase): + def setUp(self): + self._saved_keys = dict(rest_api.SERVER_PUBLIC_KEYS) + from keeper_secrets_manager_core.keeper_globals import keeper_public_keys + self._ksm_keys = keeper_public_keys + self._saved_ksm_keys = dict(keeper_public_keys) + rest_api._server_key_fetch_attempted.clear() + self.context = RestApiContext(server='https://smk.example.test') + self.context.disable_qrc() + + def tearDown(self): + rest_api.SERVER_PUBLIC_KEYS.clear() + rest_api.SERVER_PUBLIC_KEYS.update(self._saved_keys) + self._ksm_keys.clear() + self._ksm_keys.update(self._saved_ksm_keys) + rest_api._server_key_fetch_attempted.clear() + + def test_registers_the_published_key_in_commander_and_the_ksm_sdk(self): + _, key_b64 = _new_server_key() + body = {'serverPublicKeyId': 7, 'serverPublicKey': key_b64, 'fingerprint': 'SHA256:x'} + with mock.patch('requests.get', return_value=_response(200, body)) as get: + self.assertEqual(rest_api._fetch_server_public_key(self.context), 7) + get.assert_called_once() + self.assertEqual(get.call_args[0][0], 'https://smk.example.test/api/rest/server_public_key') + self.assertEqual(utils.base64_url_encode(crypto.unload_ec_public_key(rest_api.SERVER_PUBLIC_KEYS[7])), key_b64) + self.assertEqual(self._ksm_keys['7'], key_b64) + + def test_a_server_without_the_endpoint_changes_nothing(self): + before = rest_api.SERVER_PUBLIC_KEYS[7] + for rs in (_response(404, {'error': 'not_found'}), _response(200, None, 'text/html')): + rest_api._server_key_fetch_attempted.clear() + rs.json.side_effect = None if rs.status_code == 404 else ValueError('not json') + with mock.patch('requests.get', return_value=rs): + self.assertIsNone(rest_api._fetch_server_public_key(self.context)) + self.assertIs(rest_api.SERVER_PUBLIC_KEYS[7], before) + + def test_network_errors_and_bad_keys_are_swallowed(self): + before = rest_api.SERVER_PUBLIC_KEYS[7] + with mock.patch('requests.get', side_effect=OSError('connection refused')): + self.assertIsNone(rest_api._fetch_server_public_key(self.context)) + rest_api._server_key_fetch_attempted.clear() + bad = {'serverPublicKeyId': 7, 'serverPublicKey': 'not-a-key'} + with mock.patch('requests.get', return_value=_response(200, bad)): + self.assertIsNone(rest_api._fetch_server_public_key(self.context)) + self.assertIs(rest_api.SERVER_PUBLIC_KEYS[7], before) + + def test_asks_each_server_once(self): + with mock.patch('requests.get', return_value=_response(404, {})) as get: + rest_api._fetch_server_public_key(self.context) + rest_api._fetch_server_public_key(self.context) + get.assert_called_once() + + def test_execute_rest_retries_with_the_fetched_key(self): + private_key, key_b64 = _new_server_key() + server_public = crypto.load_ec_public_key(utils.base64_url_decode(key_b64)) + + def post(url, data, **kwargs): + # Answer like a server whose own key is private_key: reject anything else with key_id -1. + rq = proto.ApiRequest() + rq.ParseFromString(data) + if rest_api.SERVER_PUBLIC_KEYS.get(rq.publicKeyId) is None or \ + crypto.unload_ec_public_key(rest_api.SERVER_PUBLIC_KEYS[rq.publicKeyId]) != \ + crypto.unload_ec_public_key(server_public): + return _response(401, [{'key_id': -1, 'error': 'key', 'message': 'key'}]) + transmission_key = crypto.decrypt_ec(rq.encryptedTransmissionKey, private_key) + return _response(200, content=crypto.encrypt_aes_v2(b'ok', transmission_key), + content_type='application/octet-stream') + + body = {'serverPublicKeyId': 7, 'serverPublicKey': key_b64} + with mock.patch('requests.post', side_effect=post), \ + mock.patch('requests.get', return_value=_response(200, body)) as get: + rs = rest_api.execute_rest(self.context, 'authentication/start_login', proto.ApiRequestPayload()) + self.assertEqual(rs, b'ok') + get.assert_called_once() + + def test_keeper_cloud_never_calls_the_endpoint(self): + def post(url, data, **kwargs): + return _response(200, content=b'', content_type='application/octet-stream') + + with mock.patch('requests.post', side_effect=post), mock.patch('requests.get') as get: + rest_api.execute_rest(self.context, 'authentication/start_login', proto.ApiRequestPayload()) + get.assert_not_called() + + +if __name__ == '__main__': + unittest.main()