114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
import hmac
|
|
import hashlib
|
|
from typing import Dict, Any, Optional
|
|
from app.models.auth.module_environment_model import ModuleEnvironment
|
|
from fastapi import HTTPException, status
|
|
|
|
class TrustService:
|
|
@staticmethod
|
|
def verify_request_signature(environment: ModuleEnvironment, signature: str, payload: str = "") -> bool:
|
|
"""
|
|
Verify the HMAC signature of an incoming request from a module.
|
|
Currently supports HMAC-SHA256.
|
|
"""
|
|
if environment.trust_type != "hmac":
|
|
if environment.trust_type == "static_key":
|
|
secret = environment.trust_credentials.get("secret_key")
|
|
return hmac.compare_digest(signature, secret)
|
|
return False
|
|
|
|
secret = environment.trust_credentials.get("hmac_secret")
|
|
if not secret:
|
|
return False
|
|
|
|
expected_signature = hmac.new(
|
|
secret.encode(),
|
|
payload.encode(),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
return hmac.compare_digest(expected_signature, signature)
|
|
|
|
@staticmethod
|
|
def sign_payload(environment: ModuleEnvironment, payload: str) -> str:
|
|
"""
|
|
Signs a raw payload string using the environment's HMAC secret.
|
|
Used for direct signed POST flows.
|
|
"""
|
|
secret = environment.trust_credentials.get("hmac_secret")
|
|
if not secret:
|
|
raise ValueError(f"Module environment {environment.slug} missing 'hmac_secret'")
|
|
|
|
signature = hmac.new(
|
|
secret.encode("utf-8"),
|
|
payload.encode("utf-8"),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
return signature
|
|
|
|
@staticmethod
|
|
def sign_outbound_payload(environment: ModuleEnvironment, method: str, path: str, payload_json: str, timestamp: str) -> str:
|
|
"""
|
|
Generates HMAC-SHA256 signature for outbound requests to modules.
|
|
Signature = HMAC-SHA256(secret, method + path + timestamp + SHA256(payload))
|
|
"""
|
|
if environment.trust_type != "hmac":
|
|
pass
|
|
|
|
secret = environment.trust_credentials.get("hmac_secret")
|
|
if not secret:
|
|
raise ValueError(f"Module environment {environment.slug} missing 'hmac_secret' for outbound signing")
|
|
|
|
payload_hash = hashlib.sha256(payload_json.encode("utf-8")).hexdigest()
|
|
|
|
string_to_sign = f"{method.upper()}{path}{timestamp}{payload_hash}"
|
|
|
|
signature = hmac.new(
|
|
secret.encode("utf-8"),
|
|
string_to_sign.encode("utf-8"),
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
return signature
|
|
|
|
@staticmethod
|
|
def validate_module_trust(environment: ModuleEnvironment, request_headers: Dict[str, str], request_body: str = ""):
|
|
"""
|
|
Validates that the request comes from a trusted module environment.
|
|
Raises HTTPException if authentication fails.
|
|
"""
|
|
if not environment.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Module environment is inactive"
|
|
)
|
|
|
|
if environment.trust_type == "hmac":
|
|
signature = request_headers.get("X-Module-Signature")
|
|
if not signature:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Missing signature"
|
|
)
|
|
|
|
if not TrustService.verify_request_signature(environment, signature, request_body):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid signature"
|
|
)
|
|
|
|
elif environment.trust_type == "static_key":
|
|
api_key = request_headers.get("X-Module-Key")
|
|
secret = environment.trust_credentials.get("secret_key")
|
|
if not api_key or not hmac.compare_digest(api_key, secret):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid API Key"
|
|
)
|
|
|
|
else:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
|
detail=f"Trust type {environment.trust_type} not supported yet"
|
|
) |