Coverage for lambda / secret-rotation / handler.py: 100.00%

1""" 

2Secrets Manager rotation Lambda for the GCO backend HMAC signing key. 

3 

4This Lambda handles the four-step rotation protocol for the API Gateway 

5proxy-to-backend signing key: 

61. createSecret - Generate a new random key and store it as AWSPENDING 

72. setSecret - No-op (no external system to update) 

83. testSecret - Validate the pending key's structure 

94. finishSecret - Move AWSPENDING to AWSCURRENT 

10 

11The key is a cryptographically random string used only to compute and verify 

12request-bound HMAC envelopes; it is never transmitted to the backend. 

13Multi-region replication distributes new versions automatically. Proxies and 

14services accept both AWSCURRENT and AWSPENDING during the rotation window. 

15""" 

16 

17import json 

18import logging 

19import secrets 

20import string 

21from typing import Any 

22 

23import boto3 

24 

25# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

26# Generated at (UTC): 2026-09-01T14:42:56Z 

27# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc 

28# Flowchart(s) generated from this file: 

29# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/secret-rotation/handler.lambda_handler.html`` 

30# (PNG: ``diagrams/code_diagrams/lambda/secret-rotation/handler.lambda_handler.png``) 

31# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

32# <pyflowchart-code-diagram> END 

33 

34 

35logger = logging.getLogger() 

36logger.setLevel(logging.INFO) 

37 

38# Signing-key configuration 

39TOKEN_LENGTH = 64 

40# Alphanumeric output keeps the existing secret JSON schema and avoids escaping. 

41TOKEN_ALPHABET = string.ascii_letters + string.digits 

42 

43 

44def lambda_handler(event: dict[str, Any], context: Any) -> None: 

45 """ 

46 Handle Secrets Manager rotation request. 

47 

48 Args: 

49 event: Rotation event with Step, SecretId, ClientRequestToken 

50 context: Lambda context (unused) 

51 

52 Raises: 

53 ValueError: If the rotation step is invalid 

54 """ 

55 secret_id = event["SecretId"] 

56 token = event["ClientRequestToken"] 

57 step = event["Step"] 

58 

59 logger.info(f"Rotation step '{step}' for secret {secret_id}") 

60 

61 client = boto3.client("secretsmanager") 

62 

63 if step == "createSecret": 

64 create_secret(client, secret_id, token) 

65 elif step == "setSecret": 

66 set_secret(client, secret_id, token) 

67 elif step == "testSecret": 

68 test_secret(client, secret_id, token) 

69 elif step == "finishSecret": 

70 finish_secret(client, secret_id, token) 

71 else: 

72 raise ValueError(f"Invalid rotation step: {step}") 

73 

74 

75def create_secret(client: Any, secret_id: str, token: str) -> None: 

76 """ 

77 Create a new secret version with AWSPENDING staging label. 

78 

79 Generates a cryptographically secure random token and stores it 

80 as the pending version of the secret. 

81 

82 Args: 

83 client: Secrets Manager boto3 client 

84 secret_id: ARN or name of the secret 

85 token: Client request token for idempotency 

86 """ 

87 # Check if this version already exists (idempotency) 

88 try: 

89 client.get_secret_value(SecretId=secret_id, VersionId=token, VersionStage="AWSPENDING") 

90 logger.info(f"Secret version {token} already exists as AWSPENDING") 

91 return 

92 except client.exceptions.ResourceNotFoundException: 

93 pass # Expected - version doesn't exist yet 

94 

95 # Generate new secure random token 

96 new_token = "".join(secrets.choice(TOKEN_ALPHABET) for _ in range(TOKEN_LENGTH)) 

97 

98 # Create the secret value structure (matching original format) 

99 secret_value = json.dumps( 

100 { 

101 "description": "GCO backend HMAC signing key", 

102 "token": new_token, 

103 } 

104 ) 

105 

106 # Store as AWSPENDING 

107 client.put_secret_value( 

108 SecretId=secret_id, 

109 ClientRequestToken=token, 

110 SecretString=secret_value, 

111 VersionStages=["AWSPENDING"], 

112 ) 

113 

114 logger.info(f"Created new secret version {token} as AWSPENDING") 

115 

116 

117def set_secret(client: Any, secret_id: str, token: str) -> None: 

118 """ 

119 Set the secret in the target system. 

120 

121 For GCO's HMAC signing key, there is no external system to update. 

122 Proxies and services read staged versions directly from Secrets Manager. 

123 

124 Args: 

125 client: Secrets Manager boto3 client 

126 secret_id: ARN or name of the secret 

127 token: Client request token for idempotency 

128 """ 

129 # No-op: No external system to update 

130 # Services read directly from Secrets Manager and validate both versions 

131 logger.info("setSecret: No external system to update") 

132 

133 

134def test_secret(client: Any, secret_id: str, token: str) -> None: 

135 """ 

136 Test that the pending secret is valid. 

137 

138 For GCO's signing key, verify that the staged value can be retrieved 

139 and has the expected structure and length. 

140 

141 Args: 

142 client: Secrets Manager boto3 client 

143 secret_id: ARN or name of the secret 

144 token: Client request token for idempotency 

145 """ 

146 # Verify the pending secret can be retrieved and parsed 

147 response = client.get_secret_value( 

148 SecretId=secret_id, 

149 VersionId=token, 

150 VersionStage="AWSPENDING", 

151 ) 

152 

153 secret_data = json.loads(response["SecretString"]) 

154 

155 if "token" not in secret_data: 

156 raise ValueError("Pending secret missing 'token' field") 

157 

158 if len(secret_data["token"]) != TOKEN_LENGTH: 

159 raise ValueError(f"Token length mismatch: expected {TOKEN_LENGTH}") 

160 

161 logger.info("testSecret: Pending secret validated successfully") 

162 

163 

164def finish_secret(client: Any, secret_id: str, token: str) -> None: 

165 """ 

166 Finish the rotation by moving AWSPENDING to AWSCURRENT. 

167 

168 This atomically updates the staging labels so that: 

169 - The new version becomes AWSCURRENT 

170 - The old version loses AWSCURRENT (but may retain AWSPREVIOUS) 

171 

172 Args: 

173 client: Secrets Manager boto3 client 

174 secret_id: ARN or name of the secret 

175 token: Client request token for idempotency 

176 """ 

177 # Get current version info 

178 metadata = client.describe_secret(SecretId=secret_id) 

179 

180 # Find the current version 

181 current_version = None 

182 for version_id, stages in metadata.get("VersionIdsToStages", {}).items(): 

183 if "AWSCURRENT" in stages: 

184 if version_id == token: 

185 # Already current - rotation already completed 

186 logger.info(f"Version {token} is already AWSCURRENT") 

187 return 

188 current_version = version_id 

189 break 

190 

191 # Move AWSPENDING to AWSCURRENT 

192 client.update_secret_version_stage( 

193 SecretId=secret_id, 

194 VersionStage="AWSCURRENT", 

195 MoveToVersionId=token, 

196 RemoveFromVersionId=current_version, 

197 ) 

198 

199 logger.info(f"Rotation complete: {token} is now AWSCURRENT (was {current_version})")