(params: KeeperParams, endpoint: str, payload: Optional[bytes])
| 732 | |
| 733 | |
| 734 | def execute_router_rest(params: KeeperParams, endpoint: str, payload: Optional[bytes]) -> Optional[bytes]: |
| 735 | transmission_key = utils.generate_aes_key() |
| 736 | encrypted_session_token = crypto.encrypt_aes_v2(utils.base64_url_decode(params.session_token), transmission_key) |
| 737 | encrypted_transmission_key = rest_api.encrypt_with_keeper_key(params.rest_context, transmission_key) |
| 738 | |
| 739 | headers = { |
| 740 | 'TransmissionKey': base64.b64encode(encrypted_transmission_key).decode('ascii'), |
| 741 | 'Authorization': 'KeeperUser ' + base64.b64encode(encrypted_session_token).decode('ascii'), |
| 742 | } |
| 743 | |
| 744 | if 'ROUTER_URL' in os.environ: |
| 745 | up = urlparse(os.environ['ROUTER_URL']) |
| 746 | url_comp = (up.scheme, up.netloc, f'api/user/{endpoint}', None, None, None) |
| 747 | else: |
| 748 | from .constants import get_router_host |
| 749 | up = urlparse(params.rest_context.server_base) |
| 750 | router_host = get_router_host(up.hostname) |
| 751 | url_comp = ('https', router_host, f'api/user/{endpoint}', None, None, None) |
| 752 | url = urlunparse(url_comp) |
| 753 | |
| 754 | if payload is not None: |
| 755 | payload = crypto.encrypt_aes_v2(payload, transmission_key) |
| 756 | rs = requests.post(url, data=payload, headers=headers, proxies=params.rest_context.proxies, |
| 757 | verify=params.ssl_verify) |
| 758 | if rs.status_code == 200: |
| 759 | rs_body = rs.content |
| 760 | if rs_body: |
| 761 | router_response = router_pb2.RouterResponse() |
| 762 | router_response.ParseFromString(rs_body) |
| 763 | if router_response.responseCode == router_pb2.RouterResponseCode.RRC_OK: |
| 764 | if router_response.encryptedPayload: |
| 765 | return crypto.decrypt_aes_v2(router_response.encryptedPayload, transmission_key) |
| 766 | else: |
| 767 | if router_response.responseCode == router_pb2.RouterResponseCode.RRC_BAD_REQUEST: |
| 768 | code = 'bad_request' |
| 769 | elif router_response.responseCode == router_pb2.RouterResponseCode.RRC_NOT_ALLOWED: |
| 770 | code = 'not_allowed' |
| 771 | else: |
| 772 | code = 'router_error' |
| 773 | raise KeeperApiError(code, router_response.errorMessage) |
| 774 | else: |
| 775 | message = rs.reason |
| 776 | raise KeeperApiError('router_error', f'{message}: {rs.status_code}') |
| 777 | |
| 778 | |
| 779 | def execute_router(params, endpoint, request, *, rs_type=None): |
no test coverage detected