[BUG] Custom Auth caching ignores scopes

[huwcbjones]

Checklist

• [x] I checked the FAQ section of the documentation
• [x] I looked for similar issues in the issue tracker

Describe the bug When implementing custom auth, I'm using the scopes provided by the security schema to request an access token with the required scopes. When a request to a different endpoint that requires different scopes is made, schemathesis uses the token with the wrong scopes.

To Reproduce Steps to reproduce the behavior:

1. Add a hook implementing custom auth that looks like the sample below

@schemathesis.auth()
class OAuth2Bearer:

    @staticmethod
    def get_scopes_from_ctx(context: AuthContext) -> frozenset[str, ...] | None:
        security = context.operation.definition.get("security", [])
        if not security:
            return None
        scopes = security[0][context.operation.get_security_requirements()[0]]
        if not scopes:
            return None
        return frozenset(scopes)

    def get(self, context: AuthContext) -> str | None:
        if not (scopes := self.get_scopes_from_ctx(context)):
            return None
        token_endpoint = f"{context.operation.base_url}{TOKEN_ENDPOINT}"
        # request token with required scopes for context
        response = requests.post(token_endpoint, data={"scopes": scopes, ...})
        data = response.json()
        assert response.status_code == 200, data
        return data["access_token"]

    def set(self, case: Case, data: str, context: AuthContext) -> None:
        case.headers = case.headers or {}
        if not data:
            return
        case.headers["Authorization"] = f"Bearer {data}"

2. Run schemathesis on a schema with 2 endpoints that have different security scopes

openapi: 3.0.3
info:
  title: Cats Schema
  description: Cats Schema
  version: 1.0.0
  contact:
    email: [email protected]
servers:
  - url: https://cats.cat
tags:
  - name: cat
paths:
  /v1/cats:
    get:
      description: List cats
      operationId: listCats
      tags:
        - cat
      responses:
        "200":
          description: List of cats
          content:
            application/json:
              schema:
                type: object
                properties:
                  name:
                    type: string
      security:
        - oAuthBearer: ["list"]
    post:
      description: Create a a cat
      operationId: createCat
      tags:
        - cat
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/Cat'
      responses:
        "200":
          description: Newly created cat
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Cat'
      security:
        - oAuthBearer: ["create"]

components:
  schemas:
    Cat:
      required:
        - name
      type: object
      properties:
        name:
          type: string
          example: Macavity
  securitySchemes:
    oAuthBearer:
      type: oauth2
      flows:
        clientCredentials:
          tokenUrl: /oauth/token
          scopes:
            list: "List cats"
            create: "Create cats"

Expected behavior As documented in the schema, the different endpoints have different security scopes. Therefore, I'd expect the custom auth implementation to be called again with the different authentication context to request an access token with the right scopes.

Environment (please complete the following information): N/A, but will provide anyway

• OS: Linux or macOS
• Python version: 3.9.2
• Schemathesis version: 3.19.7
• Spec version: 3.0.3

Additional context I've bodged around it by returning the scopes from Auth.get, and comparing the scopes in the Auth.set context to the ones for the token. This obviously means that every request that requires different scopes to the original access token will end up hitting the authentication API, but it means I stop getting 403s from my application server!

    def get(self, context: AuthContext) -> _AuthData:
        # ommitted for brevity
        response = requests.post(...)
        data = response.json()
        assert response.status_code == 200, data
        return scopes, data["access_token"]

    def set(self, case: Case, data: _AuthData, context: AuthContext) -> None:
        case.headers = case.headers or {}
        if not data:
            return
        scopes, access_token = data
        required_scopes = self.get_scopes_from_ctx(context)
        if not required_scopes.issubset(scopes):
            scopes, access_token = self.get(context)
        case.headers["Authorization"] = f"Bearer {access_token}"

[Stranger6667]

Thank you so much for opening this and providing such a detailed report! :) I think we need to support custom keys in the cache implementation + check for the cached value based on the "is subset" relation for scopes.

[huwcbjones]

No worries! Your suggestion sounds reasonable 😊 If I get some spare time, I'll see if I can implement that.

[Stranger6667]

Hey @huwcbjones

In the next release, it will be possible to specify a cache key function. Something like this (though the cache key value could be anything hashable):

def extract_scopes(context):
    security = context.operation.definition.raw.get("security", [])
    if not security:
        return None
    scopes = security[0][context.operation.get_security_requirements()[0]]
    if not scopes:
        return None
    return frozenset(scopes)


@schema.auth(cache_by_key=extract_scopes)
class OAuth2Bearer:
    ...

I hope it will be helpful for your use case. Let me know if it is not enough