Deny access to certain methods based on a header.
| 800 | |
| 801 | |
| 802 | class SelectiveAuthServerMiddlewareFactory(ServerMiddlewareFactory): |
| 803 | """Deny access to certain methods based on a header.""" |
| 804 | |
| 805 | def start_call(self, info, headers): |
| 806 | if info.method == flight.FlightMethod.LIST_ACTIONS: |
| 807 | # No auth needed |
| 808 | return |
| 809 | |
| 810 | token = headers.get("x-auth-token") |
| 811 | if not token: |
| 812 | raise flight.FlightUnauthenticatedError("No token") |
| 813 | |
| 814 | token = token[0] |
| 815 | if token != "password": |
| 816 | raise flight.FlightUnauthenticatedError("Invalid token") |
| 817 | |
| 818 | return HeaderServerMiddleware(token) |
| 819 | |
| 820 | |
| 821 | class SelectiveAuthClientMiddlewareFactory(ClientMiddlewareFactory): |
no outgoing calls