This commit is contained in:
@@ -0,0 +1,71 @@
|
||||
"""DEVREQ1 call helper. Keys stay in a user-specified local file, never argv.
|
||||
|
||||
Prints signed headers by default; --send explicitly submits one HTTPS request.
|
||||
The caller keeps request-id + original body on retries and changes nonce/time.
|
||||
"""
|
||||
import argparse
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import secrets
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def sign(key, method, path, device_id, key_version, timestamp, nonce, request_id, body):
|
||||
canonical = '\n'.join(['DEVREQ1', method.upper(), path, device_id, str(key_version), str(timestamp), nonce, request_id, hashlib.sha256(body).hexdigest()])
|
||||
return hmac.new(key, canonical.encode('utf-8'), hashlib.sha256).hexdigest(), canonical
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('--device-id', required=True)
|
||||
parser.add_argument('--key-file', type=Path, required=True, help='file containing Base64URL (unpadded) encoded raw32 key')
|
||||
parser.add_argument('--key-version', type=int, default=1)
|
||||
parser.add_argument('--path', required=True)
|
||||
parser.add_argument('--body-file', type=Path, required=True)
|
||||
parser.add_argument('--request-id', required=True)
|
||||
parser.add_argument('--method', default='POST', choices=['POST', 'GET'])
|
||||
parser.add_argument('--send', help='explicit HTTPS origin, e.g. https://saas.example')
|
||||
args = parser.parse_args()
|
||||
if not args.path.startswith('/') or '?' in args.path or '#' in args.path or '\n' in args.path:
|
||||
parser.error('path must be absolute without query or fragment')
|
||||
if str(uuid.UUID(args.request_id)) != args.request_id or uuid.UUID(args.request_id).version != 4:
|
||||
parser.error('request-id must be canonical UUID v4')
|
||||
encoded = args.key_file.read_text(encoding='utf-8').strip()
|
||||
key = base64.urlsafe_b64decode(encoded + '=' * (-len(encoded) % 4))
|
||||
if len(key) != 32:
|
||||
parser.error('device key must decode to 32 bytes')
|
||||
body = args.body_file.read_bytes()
|
||||
if len(body) > 8192 or body.startswith(b'\xef\xbb\xbf'):
|
||||
parser.error('body must be <=8KiB UTF-8 without BOM')
|
||||
json.loads(body.decode('utf-8'))
|
||||
nonce = base64.urlsafe_b64encode(secrets.token_bytes(16)).decode().rstrip('=')
|
||||
timestamp = int(time.time() * 1000)
|
||||
signature, _ = sign(key, args.method, args.path, args.device_id, args.key_version, timestamp, nonce, args.request_id, body)
|
||||
headers = {'Authorization': 'Device-HMAC-SHA256 ' + signature, 'X-Device-ID': args.device_id, 'X-Key-Version': str(args.key_version), 'X-Timestamp-Ms': str(timestamp), 'X-Nonce': nonce, 'X-Request-ID': args.request_id, 'Content-Type': 'application/json'}
|
||||
if not args.send:
|
||||
print(json.dumps({'method': args.method, 'path': args.path, 'headers': headers}, ensure_ascii=False, indent=2))
|
||||
return
|
||||
origin = urllib.parse.urlsplit(args.send)
|
||||
if origin.scheme != 'https' or not origin.netloc or origin.username or origin.query or origin.fragment or origin.path not in ('', '/'):
|
||||
parser.error('--send must be a fixed HTTPS origin')
|
||||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, *unused):
|
||||
return None
|
||||
request = urllib.request.Request(args.send.rstrip('/') + args.path, data=body, headers=headers, method=args.method)
|
||||
try:
|
||||
with urllib.request.build_opener(NoRedirect).open(request, timeout=30) as response:
|
||||
print(response.read(65536).decode('utf-8'))
|
||||
except urllib.error.HTTPError as error:
|
||||
print(error.read(65536).decode('utf-8'))
|
||||
raise SystemExit(error.code)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user