Coverage for agentos/tools/url_signer.py: 0%

1""" 

2URLSigner — HMAC-based signed URL generation and verification. 

3 

4Supports: 

5 - Sign URLs with HMAC-SHA256 

6 - Expiry-based signed URLs 

7 - Path-based signature 

8 - Verification of signed URLs 

9 - Multiple signing algorithms (HS256, HS384, HS512) 

10""" 

11 

12from __future__ import annotations 

13 

14import hashlib 

15import hmac 

16import time 

17import urllib.parse 

18 

19# ============================================================================ 

20# URLSigner 

21# ============================================================================ 

22 

23 

24class URLSigner: 

25 """HMAC-based URL signing for secure temporary links. 

26 

27 Usage: 

28 signer = URLSigner(secret="my-secret-key") 

29 

30 # Generate a signed URL that expires in 1 hour 

31 signed = signer.sign("https://example.com/files/report.pdf", ttl=3600) 

32 # → https://example.com/files/report.pdf?sig=...&exp=... 

33 

34 # Verify a signed URL 

35 ok, path = signer.verify(signed) 

36 if ok: 

37 serve(path) 

38 """ 

39 

40 def __init__(self, secret: str, algorithm: str = "HS256"): 

41 self._secret = secret.encode("utf-8") 

42 algorithms = {"HS256": hashlib.sha256, "HS384": hashlib.sha384, "HS512": hashlib.sha512} 

43 if algorithm not in algorithms: 

44 raise ValueError(f"Unsupported algorithm: {algorithm}. Use HS256/HS384/HS512") 

45 self._hash_func = algorithms[algorithm] 

46 self._algorithm = algorithm 

47 

48 def sign( 

49 self, 

50 url: str, 

51 ttl: float | None = None, 

52 extra_params: dict | None = None, 

53 ) -> str: 

54 """Sign a URL with optional TTL and extra params. 

55 

56 Args: 

57 url: The URL to sign 

58 ttl: Time-to-live in seconds. None = no expiry 

59 extra_params: Additional query params to include in signature 

60 """ 

61 parsed = urllib.parse.urlparse(url) 

62 params = dict(urllib.parse.parse_qsl(parsed.query)) 

63 

64 # Build signature payload 

65 path = parsed.path 

66 if extra_params: 

67 for k, v in sorted(extra_params.items()): 

68 params[k] = str(v) 

69 

70 if ttl is not None: 

71 exp = int(time.time() + ttl) 

72 params["exp"] = str(exp) 

73 

74 # Generate signature over path + sorted params 

75 sig = self._compute_signature(path, params) 

76 

77 params["sig"] = sig 

78 

79 new_query = urllib.parse.urlencode(params) 

80 return urllib.parse.urlunparse(parsed._replace(query=new_query)) 

81 

82 def verify(self, url: str) -> tuple[bool, str | None]: 

83 """Verify a signed URL. Returns (is_valid, error_message).""" 

84 parsed = urllib.parse.urlparse(url) 

85 params = dict(urllib.parse.parse_qsl(parsed.query)) 

86 

87 sig = params.pop("sig", None) 

88 if not sig: 

89 return False, "Missing signature" 

90 

91 # Check expiry (do not pop — must remain for signature recalculation) 

92 exp = params.get("exp") 

93 if exp: 

94 exp_val = int(exp) 

95 if time.time() > exp_val: 

96 return ( 

97 False, 

98 f"URL expired at {time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(exp_val))}", 

99 ) 

100 

101 # Recompute 

102 path = parsed.path 

103 expected = self._compute_signature(path, params) 

104 

105 if not hmac.compare_digest(sig, expected): 

106 return False, "Invalid signature" 

107 

108 return True, None 

109 

110 def _compute_signature(self, path: str, params: dict) -> str: 

111 data = path.encode("utf-8") 

112 for k in sorted(params.keys()): 

113 v = params[k] 

114 data += f"|{k}={v}".encode() 

115 return hmac.new(self._secret, data, self._hash_func).hexdigest()