如何实现对响应生效的Auth处理器？OAuth认证POST重定向问题

Hey there, let's break down why your current approach is failing and how to fix it properly.

The Root Problem #

Your current implementation relies on response hooks to handle Keycloak redirects, but two critical issues are breaking POST requests:

1. Request method mismatch on redirect: When requests follows a 3xx redirect from a POST request, it may retain the POST method (depending on the status code) and send it to Keycloak's login endpoint—but Keycloak only accepts GET requests to serve the login form.
2. Hooks don't run on internal redirects: Requests handles redirects internally, so your response hook only fires for the initial request's response. The redirect requests themselves skip the hook entirely, so you can't modify their method there.

Solution: Handle Authentication Proactively in the Auth Handler #

Instead of waiting for a redirect and trying to patch it with hooks, we can rewrite the KeycloakAuth class to handle the full Keycloak authentication flow upfront before sending the target request. This works for both GET and POST requests because we avoid the problematic redirect method mismatch entirely.

Approach Overview #

1. In the __call__ method (which requests calls to apply auth), first try sending the request with any existing auth credentials.
2. If we get a 401 or a redirect to Keycloak, trigger the authentication flow manually:
   • Send a GET request to Keycloak's login page to grab required form parameters (like CSRF tokens and execution IDs).
   • Submit the login credentials to Keycloak's form endpoint.
   • Save the authenticated session cookies/tokens.
3. Resend the original request with the valid auth credentials.

Example Implementation #

import requests
from bs4 import BeautifulSoup

class KeycloakAuth(requests.auth.AuthBase):
    def __init__(self, server_url, realm, client_id, username, password):
        self.server_url = server_url
        self.realm = realm
        self.client_id = client_id
        self.username = username
        self.password = password
        self.session = requests.Session()
        self.auth_cookies = {}

    def __call__(self, request):
        # Attach existing auth cookies if we have them
        request.cookies.update(self.auth_cookies)
        
        # Send initial request without following redirects
        response = self.session.send(request, allow_redirects=False)
        
        # Check if we need to authenticate
        if (response.status_code in (302, 401) 
            and 'keycloak' in response.headers.get('Location', '')):
            # Complete Keycloak authentication flow
            self._authenticate(response.headers['Location'])
            # Update the request with fresh auth cookies
            request.cookies.update(self.auth_cookies)
            # Resend the request with valid auth
            response = self.session.send(request)
        
        return response

    def _authenticate(self, login_url):
        # Step 1: Fetch login page to get form parameters
        login_page = self.session.get(login_url)
        soup = BeautifulSoup(login_page.text, 'html.parser')
        
        # Extract required form fields
        csrf_token = soup.find('input', {'name': 'csrf_token'})['value']
        execution = soup.find('input', {'name': 'execution'})['value']
        form_action = soup.find('form')['action']

        # Step 2: Submit login credentials
        login_payload = {
            'username': self.username,
            'password': self.password,
            'csrf_token': csrf_token,
            'execution': execution,
            'client_id': self.client_id,
            'submit': 'Login'
        }
        
        # Follow redirects after login to capture auth cookies
        auth_response = self.session.post(form_action, data=login_payload, allow_redirects=True)
        self.auth_cookies = auth_response.cookies.get_dict()

# Usage Example
if __name__ == "__main__":
    keycloak_auth = KeycloakAuth(
        server_url="https://your-keycloak-instance/auth",
        realm="your-realm-name",
        client_id="your-client-id",
        username="your-username",
        password="your-password"
    )

    # Works with GET
    get_response = requests.get("https://your-protected-api/resource", auth=keycloak_auth)
    # Works with POST too!
    post_response = requests.post(
        "https://your-protected-api/resource",
        json={"data": "test-payload"},
        auth=keycloak_auth
    )

Key Improvements #

• Manual redirect control: By setting allow_redirects=False, we prevent requests from automatically sending a POST to Keycloak's login endpoint.
• Proactive auth handling: We handle the full login flow in the auth handler, so both GET and POST requests get valid auth before hitting the protected resource.
• Session persistence: Using requests.Session keeps cookies across requests, so we don't re-authenticate on every call.

Bonus: Use Password Grant Flow (If Allowed) #

If your Keycloak client is configured to allow the OAuth2 Password Grant (best suited for backend services, not user-facing apps), you can skip form-based login entirely and fetch a token directly via API. This is more reliable and simpler:

import requests

class KeycloakAuth(requests.auth.AuthBase):
    def __init__(self, server_url, realm, client_id, username, password):
        self.token_endpoint = f"{server_url}/realms/{realm}/protocol/openid-connect/token"
        self.client_id = client_id
        self.username = username
        self.password = password
        self.access_token = None

    def _fetch_token(self):
        token_payload = {
            "grant_type": "password",
            "client_id": self.client_id,
            "username": self.username,
            "password": self.password
        }
        response = requests.post(self.token_endpoint, data=token_payload)
        response.raise_for_status()
        self.access_token = response.json()["access_token"]

    def __call__(self, request):
        if not self.access_token:
            self._fetch_token()
        # Add Bearer token to request headers
        request.headers["Authorization"] = f"Bearer {self.access_token}"
        return request

# Usage
auth = KeycloakAuth(
    server_url="https://your-keycloak-instance/auth",
    realm="your-realm-name",
    client_id="your-client-id",
    username="your-username",
    password="your-password"
)
requests.post("https://your-protected-api/resource", json={"data": "test"}, auth=auth)

This approach avoids dealing with form parsing and redirects entirely, but make sure the Password Grant is enabled for your Keycloak client (it's disabled by default for security reasons).

内容的提问来源于stack exchange，提问作者ackerleytng