Add simple Basic auth (#1203)

* Add simple Basic auth

To enable the basic authentication, one must set `server.auth.enabled`
to true.

The static string defined in `server.auth.secret` must be set in the
header `Authorization`.

The health check endpoint will always be accessible, no matter the API
auth configuration.

* Fix linting and type check

* Fighting with mypy being too restrictive

Had to disable mypy in the `auth` as we are not using the same signature
for the authenticated method.

mypy was complaining that the signatures of `authenticated` must be
identical, no matter in which logical branch we are.
Given that fastapi is accomodating itself of method signatures (it will
inject the dependencies in the method call), this warning of mypy is
actually preventing us to do something legit.

mypy doc: https://mypy.readthedocs.io/en/stable/common_issues.html

* Write tests to verify that the simple auth is working

private_gpt/server/chat/chat_router.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 from llama_index.llms import ChatMessage, MessageRole
 from pydantic import BaseModel
 from starlette.responses import StreamingResponse
@@ -12,8 +12,9 @@ from private_gpt.open_ai.openai_models import (
     to_openai_sse_stream,
 )
 from private_gpt.server.chat.chat_service import ChatService
+from private_gpt.server.utils.auth import authenticated
 
-chat_router = APIRouter(prefix="/v1")
+chat_router = APIRouter(prefix="/v1", dependencies=[Depends(authenticated)])
 
 
 class ChatBody(BaseModel):

private_gpt/server/chunks/chunks_router.py

@@ -1,13 +1,14 @@
 from typing import Literal
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 from pydantic import BaseModel, Field
 
 from private_gpt.di import root_injector
 from private_gpt.open_ai.extensions.context_filter import ContextFilter
 from private_gpt.server.chunks.chunks_service import Chunk, ChunksService
+from private_gpt.server.utils.auth import authenticated
 
-chunks_router = APIRouter(prefix="/v1")
+chunks_router = APIRouter(prefix="/v1", dependencies=[Depends(authenticated)])
 
 
 class ChunksBody(BaseModel):

private_gpt/server/completions/completions_router.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 from pydantic import BaseModel
 from starlette.responses import StreamingResponse
 
@@ -8,8 +8,9 @@ from private_gpt.open_ai.openai_models import (
     OpenAIMessage,
 )
 from private_gpt.server.chat.chat_router import ChatBody, chat_completion
+from private_gpt.server.utils.auth import authenticated
 
-completions_router = APIRouter(prefix="/v1")
+completions_router = APIRouter(prefix="/v1", dependencies=[Depends(authenticated)])
 
 
 class CompletionsBody(BaseModel):

private_gpt/server/embeddings/embeddings_router.py

@@ -1,6 +1,6 @@
 from typing import Literal
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 from pydantic import BaseModel
 
 from private_gpt.di import root_injector
@@ -8,8 +8,9 @@ from private_gpt.server.embeddings.embeddings_service import (
     Embedding,
     EmbeddingsService,
 )
+from private_gpt.server.utils.auth import authenticated
 
-embeddings_router = APIRouter(prefix="/v1")
+embeddings_router = APIRouter(prefix="/v1", dependencies=[Depends(authenticated)])
 
 
 class EmbeddingsBody(BaseModel):

private_gpt/server/health/health_router.py

@@ -3,6 +3,7 @@ from typing import Literal
 from fastapi import APIRouter
 from pydantic import BaseModel, Field
 
+# Not authentication or authorization required to get the health status.
 health_router = APIRouter()
 
 

private_gpt/server/ingest/ingest_router.py

@@ -1,12 +1,13 @@
 from typing import Literal
 
-from fastapi import APIRouter, HTTPException, UploadFile
+from fastapi import APIRouter, Depends, HTTPException, UploadFile
 from pydantic import BaseModel
 
 from private_gpt.di import root_injector
 from private_gpt.server.ingest.ingest_service import IngestedDoc, IngestService
+from private_gpt.server.utils.auth import authenticated
 
-ingest_router = APIRouter(prefix="/v1")
+ingest_router = APIRouter(prefix="/v1", dependencies=[Depends(authenticated)])
 
 
 class IngestResponse(BaseModel):

private_gpt/server/utils/auth.py

@@ -0,0 +1,68 @@
+"""Authentication mechanism for the API.
+
+Define a simple mechanism to authenticate requests.
+More complex authentication mechanisms can be defined here, and be placed in the
+`authenticated` method (being a 'bean' injected in fastapi routers).
+
+Authorization can also be made after the authentication, and depends on
+the authentication. Authorization should not be implemented in this file.
+
+Authorization can be done by following fastapi's guides:
+* https://fastapi.tiangolo.com/advanced/security/oauth2-scopes/
+* https://fastapi.tiangolo.com/tutorial/security/
+* https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/
+"""
+# mypy: ignore-errors
+# Disabled mypy error: All conditional function variants must have identical signatures
+# We are changing the implementation of the authenticated method, based on
+# the config. If the auth is not enabled, we are not defining the complex method
+# with its dependencies.
+import logging
+import secrets
+from typing import Annotated
+
+from fastapi import Depends, Header, HTTPException
+
+from private_gpt.settings.settings import settings
+
+# 401 signify that the request requires authentication.
+# 403 signify that the authenticated user is not authorized to perform the operation.
+NOT_AUTHENTICATED = HTTPException(
+    status_code=401,
+    detail="Not authenticated",
+    headers={"WWW-Authenticate": 'Basic realm="All the API", charset="UTF-8"'},
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _simple_authentication(authorization: Annotated[str, Header()] = "") -> bool:
+    """Check if the request is authenticated."""
+    if not secrets.compare_digest(authorization, settings.server.auth.secret):
+        # If the "Authorization" header is not the expected one, raise an exception.
+        raise NOT_AUTHENTICATED
+    return True
+
+
+if not settings.server.auth.enabled:
+    logger.debug(
+        "Defining a dummy authentication mechanism for fastapi, always authenticating requests"
+    )
+
+    # Define a dummy authentication method that always returns True.
+    def authenticated() -> bool:
+        """Check if the request is authenticated."""
+        return True
+
+else:
+    logger.info("Defining the given authentication mechanism for the API")
+
+    # Method to be used as a dependency to check if the request is authenticated.
+    def authenticated(
+        _simple_authentication: Annotated[bool, Depends(_simple_authentication)]
+    ) -> bool:
+        """Check if the request is authenticated."""
+        assert settings.server.auth.enabled
+        if not _simple_authentication:
+            raise NOT_AUTHENTICATED
+        return True