()
| 103 | |
| 104 | @contextmanager |
| 105 | def _metadata_server() -> Iterator[tuple[str, list[tuple[str, str, str | None]]]]: |
| 106 | calls: list[tuple[str, str, str | None]] = [] |
| 107 | |
| 108 | class Handler(BaseHTTPRequestHandler): |
| 109 | def _respond(self, body: str, *, content_type: str = "text/plain") -> None: |
| 110 | encoded = body.encode() |
| 111 | self.send_response(200) |
| 112 | self.send_header("Content-Type", content_type) |
| 113 | self.send_header("Content-Length", str(len(encoded))) |
| 114 | self.end_headers() |
| 115 | self.wfile.write(encoded) |
| 116 | |
| 117 | def do_PUT(self) -> None: |
| 118 | calls.append(("PUT", self.path, self.headers.get("Authorization"))) |
| 119 | if self.path != "/latest/api/token": |
| 120 | self.send_error(404) |
| 121 | return |
| 122 | self._respond("metadata-token") |
| 123 | |
| 124 | def do_GET(self) -> None: |
| 125 | calls.append(("GET", self.path, self.headers.get("Authorization"))) |
| 126 | if self.path == "/container-credentials": |
| 127 | self._respond( |
| 128 | json.dumps( |
| 129 | { |
| 130 | "AccessKeyId": "container-access-key", |
| 131 | "SecretAccessKey": "container-secret-key", |
| 132 | "Token": "container-session-token", |
| 133 | "Expiration": _FUTURE_EXPIRATION, |
| 134 | } |
| 135 | ), |
| 136 | content_type="application/json", |
| 137 | ) |
| 138 | return |
| 139 | if self.path == "/latest/meta-data/iam/security-credentials/": |
| 140 | self._respond("instance-role") |
| 141 | return |
| 142 | if self.path == "/latest/meta-data/iam/security-credentials/instance-role": |
| 143 | self._respond( |
| 144 | json.dumps( |
| 145 | { |
| 146 | "Code": "Success", |
| 147 | "LastUpdated": "2026-01-01T00:00:00Z", |
| 148 | "Type": "AWS-HMAC", |
| 149 | "AccessKeyId": "imds-access-key", |
| 150 | "SecretAccessKey": "imds-secret-key", |
| 151 | "Token": "imds-session-token", |
| 152 | "Expiration": _FUTURE_EXPIRATION, |
| 153 | } |
| 154 | ), |
| 155 | content_type="application/json", |
| 156 | ) |
| 157 | return |
| 158 | self.send_error(404) |
| 159 | |
| 160 | @override |
| 161 | def log_message(self, format: str, *args: Any) -> None: |
| 162 | del format, args |
no test coverage detected