Decode and validate a previously issued resume token. Removes standard claims before returning the decoded payload. Raises HTTP 400 on expired/invalid tokens to match previous behavior.
(token: str)
| 27 | |
| 28 | |
| 29 | def read_stream_token(token: str) -> dict[str, Any]: |
| 30 | """Decode and validate a previously issued resume token. |
| 31 | |
| 32 | Removes standard claims before returning the decoded payload. |
| 33 | Raises HTTP 400 on expired/invalid tokens to match previous behavior. |
| 34 | """ |
| 35 | try: |
| 36 | decoded: dict[str, Any] = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALG]) |
| 37 | decoded.pop("iat", None) |
| 38 | decoded.pop("exp", None) |
| 39 | return decoded |
| 40 | except jwt.ExpiredSignatureError: |
| 41 | raise HTTPException(status_code=400, detail="Token expired") |
| 42 | except jwt.InvalidTokenError: |
| 43 | raise HTTPException(status_code=400, detail="Invalid token") |
no outgoing calls
no test coverage detected