"""Python 3.10+, standard library only. Keep API secrets on your server.""" import base64 import hashlib import hmac import http.client import os import re import secrets import sys import time from urllib.parse import parse_qsl, quote_plus, urlencode, urlsplit def canonical_query(raw): if ';' in raw or re.search(r'%(?![0-9A-Fa-f]{2})', raw): raise ValueError('Invalid query encoding') pairs = parse_qsl(raw, keep_blank_values=True, encoding='utf-8', errors='strict') # Stable sort by UTF-8 key; preserve duplicate-value ordering. pairs.sort(key=lambda pair: pair[0].encode('utf-8')) return urlencode(pairs, quote_via=quote_plus, safe='~') def canonical_request(method, path, query, timestamp, nonce, body=b''): if not re.fullmatch(r"[!#$%&'*+.^_`|~0-9A-Za-z-]+", method): raise ValueError('Invalid HTTP method') if not re.fullmatch(r"/(?:[A-Za-z0-9\-._~!$&'()*+,;=:@/]|%[0-9A-Fa-f]{2})*", path): raise ValueError('Use the exact escaped path') if not re.fullmatch(r'[A-Za-z0-9_-]{22,128}', nonce): raise ValueError('Invalid nonce') if type(timestamp) is not int or timestamp < 0: raise ValueError('Invalid timestamp') return '\n'.join([method.upper(), path, canonical_query(query), str(timestamp), nonce, hashlib.sha256(body).hexdigest()]) def sign(secret, method, path, query, timestamp, nonce, body=b''): canonical = canonical_request(method, path, query, timestamp, nonce, body) # Do not base64-decode the issued secret. Use its literal UTF-8 bytes. return hmac.new(secret.encode('utf-8'), canonical.encode('utf-8'), hashlib.sha256).hexdigest() def main(): base = urlsplit(os.environ['FTM_API_BASE']) key, secret, regcode = (os.environ[k] for k in ('FTM_API_KEY', 'FTM_API_SECRET', 'FTM_REGCODE')) if not re.fullmatch(r'[0-9]{11}', regcode): raise ValueError('FTM_REGCODE must contain 11 digits') if not base.hostname or base.username or base.password or base.path not in ('', '/') or base.query or base.fragment: raise ValueError('FTM_API_BASE must be an origin without credentials, path, query or fragment') if base.scheme != 'https': raise ValueError('HTTPS required') method, path, query, body = 'GET', '/service/api/v1/legal-entities/' + regcode, '', b'' timestamp = int(time.time()) nonce = base64.urlsafe_b64encode(secrets.token_bytes(16)).rstrip(b'=').decode('ascii') headers = {'X-API-Key': key, 'X-API-Timestamp': str(timestamp), 'X-API-Nonce': nonce, 'X-API-Signature': sign(secret, method, path, query, timestamp, nonce, body), 'Accept': 'application/json'} connection = http.client.HTTPSConnection(base.hostname, base.port, timeout=30) try: # http.client sends this exact escaped target and does not follow redirects. connection.request(method, path + ('?' + query if query else ''), body=body, headers=headers) response = connection.getresponse() while chunk := response.read(65536): sys.stdout.buffer.write(chunk) sys.stdout.buffer.write(b'\n') return 0 if response.status == 200 else 1 finally: connection.close() if __name__ == '__main__': try: sys.exit(main()) except (KeyError, ValueError, OSError, http.client.HTTPException): sys.exit('Request failed. Check configuration, connectivity and API access.')