(self, bundle: dict)
| 1128 | return float(exp) if isinstance(exp, int) else float("inf") |
| 1129 | |
| 1130 | def _discover_token_endpoint(self, bundle: dict) -> str: |
| 1131 | if self._token_endpoint is not None: |
| 1132 | return self._token_endpoint |
| 1133 | issuer = bundle.get("issuer") |
| 1134 | if not isinstance(issuer, str) or not issuer: |
| 1135 | raise SandboxError( |
| 1136 | f"OIDC bundle for gateway '{self._cluster_name}' has no " |
| 1137 | f"`issuer`; cannot refresh. Re-authenticate with: openshell " |
| 1138 | f"gateway login" |
| 1139 | ) |
| 1140 | normalized_issuer = issuer.rstrip("/") |
| 1141 | discovery_url = f"{normalized_issuer}/.well-known/openid-configuration" |
| 1142 | try: |
| 1143 | resp = self._http.get(discovery_url) |
| 1144 | except httpx.HTTPError as e: |
| 1145 | raise SandboxError( |
| 1146 | f"OIDC discovery failed for gateway " |
| 1147 | f"'{self._cluster_name}': {e}. Re-authenticate with: " |
| 1148 | f"openshell gateway login" |
| 1149 | ) from e |
| 1150 | # follow_redirects=False means a 3xx surfaces as a non-2xx |
| 1151 | # status; treat any non-success as a discovery failure rather |
| 1152 | # than silently following. |
| 1153 | if not 200 <= resp.status_code < 300: |
| 1154 | raise SandboxError( |
| 1155 | f"OIDC discovery failed for gateway " |
| 1156 | f"'{self._cluster_name}': HTTP {resp.status_code} " |
| 1157 | f"from {discovery_url}. Re-authenticate with: openshell " |
| 1158 | f"gateway login" |
| 1159 | ) |
| 1160 | try: |
| 1161 | disco = resp.json() |
| 1162 | except ValueError as e: |
| 1163 | raise SandboxError( |
| 1164 | f"OIDC discovery returned invalid JSON for gateway " |
| 1165 | f"'{self._cluster_name}': {e}" |
| 1166 | ) from e |
| 1167 | # Critical: validate that the discovery document's `issuer` |
| 1168 | # matches the configured one. Without this, a misdirected or |
| 1169 | # malicious discovery response could steer the refresh_token |
| 1170 | # POST to an attacker-controlled endpoint. |
| 1171 | discovered_issuer = disco.get("issuer", "") |
| 1172 | if not isinstance(discovered_issuer, str) or ( |
| 1173 | discovered_issuer.rstrip("/") != normalized_issuer |
| 1174 | ): |
| 1175 | raise SandboxError( |
| 1176 | f"OIDC discovery issuer mismatch for gateway " |
| 1177 | f"'{self._cluster_name}': expected '{normalized_issuer}', " |
| 1178 | f"got '{discovered_issuer}'." |
| 1179 | ) |
| 1180 | endpoint = disco.get("token_endpoint") |
| 1181 | if not isinstance(endpoint, str) or not endpoint: |
| 1182 | raise SandboxError( |
| 1183 | f"OIDC discovery for gateway '{self._cluster_name}' did " |
| 1184 | f"not include a token_endpoint." |
| 1185 | ) |
| 1186 | self._token_endpoint = endpoint |
| 1187 | return endpoint |
no test coverage detected