[Python] Pysolr JWT Token can't be passed as authentication

Discussão em 'Python' iniciado por Stack, Fevereiro 25, 2025.

  1. Stack

    Stack Membro Participativo

    Structure

    • SOLR v. 9.8.0, 3 Nodes
    • Zookeeper v. 3.9.3, 3 Instances
    • Pysolr latest

    The SOLR and Zookeeper Nodes run in a docker environment.

    The Solr Authentication is a MultiAuthPlugin with the BasicAuthPlugin used to log in to the SOLR Admin UI and the JWTAuthPlugin used for the requests.

    What I want

    I need to send a request from my RestAPI to SOLR so I can search for content. For this I use pysolr.

    The Problem

    When I send the request with the token given as auth parameter I always get a "TypeError: 'str' object is not callable" error.

    What I tried

    I first tried my function without the authentication which works and returns my search results as expected. Then I tried with a static token which returns the error. And finally I tried the same with a always newly generated token which as well returns the error. Since I wasn't sure if the generated token works I printed the token in the console and tried to send a request with Postman with the given token which did work. When I use the "Basic Authentication" using a username and password it works as well.

    What I expected

    I expect, that the request is sent to SOLR. Then SOLR should check if the JWT token is valid with the settings set in the security.json and finally should return the search results.

    Code

    from typing import Any

    import jwt
    import pysolr
    from fastapi import APIRouter, Depends, HTTPException, status
    from fastapi_cache.decorator import cache

    from app.caching import TTL
    from app.core.config import settings
    from app.models import SearchResults
    from app.queryparams import SearchQueryParams
    from app.utils import check_search_input

    router = APIRouter()

    @router.get("/", response_model=SearchResults)
    # @cache(expire=TTL.WEEK, namespace="public:search")
    async def search(
    skip: int = 0,
    limit: int = 100,
    params: SearchQueryParams = Depends(),
    ) -> Any:
    """
    Search query.
    """
    if not check_search_input(params=params):
    raise HTTPException(
    status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
    detail="Mindestens ein Feld muss ausgefüllt werden",
    )

    zookeeper = pysolr.ZooKeeper("search-zoo1,search-zoo2,search-zoo3")

    jwt_token = jwt.encode({
    "sub": "test-backend",
    "name": "test-backend",
    "iat": 1516239022,
    "admin": False,
    "iss": "test-solr-auth",
    "solr": "test-backend",
    "aud": "test-solr-api",
    "exp": 33292598400,
    "scope": "test-backend",
    "realm": "test-solr-jwt"
    }, settings.KEY, algorithm="RS256")

    solr = pysolr.SolrCloud(zookeeper, "tag", auth=jwt_token)

    res = solr.search(q=params.query, start=skip, rows=limit)

    return SearchResults(data=res, count=res.hits)

    Continue reading...

Compartilhe esta Página