mohawk

import os from mohawk import Sender, Receiver from mohawk.exc import HawkAuthenticateError # --- Shared Credentials (typically stored securely) --- credentials = { 'id': os.environ.get('HAWK_ID', 'some-id'), 'key': os.environ.get('HAWK_KEY', 'a super secret key'), 'algorithm': 'sha256' } url = 'http://example.com/resource' method = 'POST' content = b'this is some test content' content_type = 'text/plain' # --- Sender (Client-side) --- def make_hawk_request(credentials, url, method, content, content_type): sender = Sender( credentials, url, method, content=content, content_type=content_type ) headers = { 'Authorization': sender.request_header, 'Content-Type': content_type } print(f"\nSender generated Authorization header: {sender.request_header}") # In a real application, you would send this via requests.post(url, headers=headers, data=content) return headers, content # --- Receiver (Server-side) --- def receive_hawk_request(credentials, request_headers, request_content, url, method): try: # The `lookup_credentials` and `seen_nonce` are application-specific callbacks # For this example, we'll use simple in-memory functions. def lookup_credentials(sender_id): if sender_id == credentials['id']: return credentials return None # In a real app, this would check a database/cache for replay attacks processed_nonces = set() def seen_nonce(sender_id, nonce, timestamp): # A simple, insecure example. DO NOT USE IN PRODUCTION. # Real implementation needs a persistent, shared, and atomic store. key = f"{sender_id}:{nonce}:{timestamp}" if key in processed_nonces: return True processed_nonces.add(key) return False receiver = Receiver( lookup_credentials, # Callback to retrieve credentials request_headers['Authorization'], # Incoming Authorization header url, # Request URL method, # Request method content=request_content, # Request body content_type=request_headers['Content-Type'], # Request Content-Type seen_nonce=seen_nonce, # Callback to check for replay attacks ) print("\nReceiver: Hawk authentication successful!") print(f"Sender ID: {receiver.credentials['id']}") print(f"Ext data: {receiver.ext}") # Optionally, the receiver can sign its response response_content = b'response from server' response_content_type = 'text/plain' receiver.respond( content=response_content, content_type=response_content_type ) print(f"Receiver generated Server-Authorization header: {receiver.response_header}") return True except HawkAuthenticateError as e: print(f"\nReceiver: Hawk authentication failed: {e}") return False # --- Simulate a request-response cycle --- request_headers, request_body = make_hawk_request(credentials, url, method, content, content_type) # Simulate server receiving and processing the request success = receive_hawk_request(credentials, request_headers, request_body, url, method) if success: print("End-to-end Hawk flow demonstrated successfully.") else: print("Hawk flow failed.")