added optional global AppAPIAuth middleware

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.8.1 - 2024-0x-xx]
+
+### Added
+
+- NextcloudApp: `AppAPIAuthMiddleware` for easy cover all endpoints. #205
+
 ## [0.8.0 - 2024-01-12]
 
 ### Added

docs/NextcloudApp.rst

@@ -223,4 +223,29 @@ and since this is not directly related to working with NextCloud, we will skip t
 
 **ToGif** example `full source <https://github.com/cloud-py-api/nc_py_api/blob/main/examples/as_app/to_gif/lib/main.py>`_ code.
 
+Using AppAPIAuthMiddleware
+--------------------------
+
+If your application does not implement `Talk Bot` functionality and you most often do not need
+the ``NextcloudApp`` class returned after standard authentication with `Depends`:
+
+.. code-block:: python
+
+    nc: Annotated[NextcloudApp, Depends(nc_app)]
+
+In this case, you can use global authentication. It's quite simple, just add this line of code:
+
+.. code-block:: python
+
+    from nc_py_api.ex_app import AppAPIAuthMiddleware
+
+    APP = FastAPI(lifespan=lifespan)
+    APP.add_middleware(AppAPIAuthMiddleware)
+
+and it will be called for all your endpoints and check the validity of the connection itself.
+
+``AppAPIAuthMiddleware`` supports **disable_for** optional argument, where you can list all routes for which authentication should be skipped.
+
+You can still use at the same time the *AppAPIAuthMiddleware* and *Depends(nc_app)*, it is clever enough and they won't interfere with each other.
+
 This chapter ends here, but the next topics are even more intriguing.

docs/conf.py

@@ -66,6 +66,7 @@
 nitpicky = True
 nitpick_ignore_regex = [
     (r"py:class", r"starlette\.requests\.Request"),
+    (r"py:class", r"starlette\.requests\.HTTPConnection"),
     (r"py:.*", r"httpx.*"),
 ]
 

nc_py_api/_session.py

@@ -11,8 +11,8 @@
 from json import loads
 from os import environ
 
-from fastapi import Request as FastAPIRequest
 from httpx import AsyncClient, Client, Headers, Limits, ReadTimeout, Request, Response
+from starlette.requests import HTTPConnection
 
 from . import options
 from ._exceptions import (
@@ -459,7 +459,7 @@ def __init__(self, **kwargs):
         self.cfg = AppConfig(**kwargs)
         super().__init__(**kwargs)
 
-    def sign_check(self, request: FastAPIRequest) -> None:
+    def sign_check(self, request: HTTPConnection) -> None:
         headers = {
             "AA-VERSION": request.headers.get("AA-VERSION", ""),
             "EX-APP-ID": request.headers.get("EX-APP-ID", ""),

nc_py_api/_version.py

@@ -1,3 +1,3 @@
 """Version of nc_py_api."""
 
-__version__ = "0.8.0"
+__version__ = "0.8.1.dev0"

nc_py_api/ex_app/__init__.py

@@ -2,6 +2,7 @@
 
 from .defs import ApiScope, LogLvl
 from .integration_fastapi import (
+    AppAPIAuthMiddleware,
     anc_app,
     atalk_bot_app,
     nc_app,

nc_py_api/ex_app/integration_fastapi.py

@@ -12,19 +12,20 @@
     Depends,
     FastAPI,
     HTTPException,
-    Request,
     responses,
     staticfiles,
     status,
 )
+from starlette.requests import HTTPConnection, Request
+from starlette.types import ASGIApp, Receive, Scope, Send
 
 from .._misc import get_username_secret_from_headers
 from ..nextcloud import AsyncNextcloudApp, NextcloudApp
 from ..talk_bot import TalkBotMessage, aget_bot_secret, get_bot_secret
 from .misc import persistent_storage
 
 
-def nc_app(request: Request) -> NextcloudApp:
+def nc_app(request: HTTPConnection) -> NextcloudApp:
     """Authentication handler for requests from Nextcloud to the application."""
     user = get_username_secret_from_headers(
         {"AUTHORIZATION-APP-API": request.headers.get("AUTHORIZATION-APP-API", "")}
@@ -36,7 +37,7 @@ def nc_app(request: Request) -> NextcloudApp:
     return nextcloud_app
 
 
-def anc_app(request: Request) -> AsyncNextcloudApp:
+def anc_app(request: HTTPConnection) -> AsyncNextcloudApp:
     """Async Authentication handler for requests from Nextcloud to the application."""
     user = get_username_secret_from_headers(
         {"AUTHORIZATION-APP-API": request.headers.get("AUTHORIZATION-APP-API", "")}
@@ -194,3 +195,40 @@ def display(self, msg=None, pos=None):
             cache = models[model].pop("cache_dir", persistent_storage())
             snapshot_download(model, tqdm_class=TqdmProgress, **models[model], max_workers=workers, cache_dir=cache)
     nc.set_init_status(100)
+
+
+class AppAPIAuthMiddleware:
+    """Pure ASGI AppAPIAuth Middleware."""
+
+    _disable_for: list[str]
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        disable_for: list[str] | None = None,
+    ) -> None:
+        self.app = app
+        disable_for = [] if disable_for is None else [i.lstrip("/") for i in disable_for]
+        self._disable_for = [i for i in disable_for if i != "heartbeat"] + ["heartbeat"]
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """Method that will be called by Starlette for each event."""
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        conn = HTTPConnection(scope)
+        url_path = conn.url.path.lstrip("/")
+        if url_path not in self._disable_for:
+            try:
+                anc_app(conn)
+            except HTTPException as exc:
+                response = self._on_error(exc.status_code, exc.detail)
+                await response(scope, receive, send)
+                return
+
+        await self.app(scope, receive, send)
+
+    @staticmethod
+    def _on_error(status_code: int = 400, content: str = "") -> responses.PlainTextResponse:
+        return responses.PlainTextResponse(content, status_code=status_code)

nc_py_api/nextcloud.py

@@ -3,8 +3,8 @@
 import typing
 from abc import ABC
 
-from fastapi import Request as FastAPIRequest
 from httpx import Headers
+from starlette.requests import HTTPConnection
 
 from ._exceptions import NextcloudExceptionNotFound
 from ._misc import check_capabilities, require_capabilities
@@ -395,7 +395,7 @@ def unregister_talk_bot(self, callback_url: str) -> bool:
             return False
         return True
 
-    def request_sign_check(self, request: FastAPIRequest) -> bool:
+    def request_sign_check(self, request: HTTPConnection) -> bool:
         """Verifies the signature and validity of an incoming request from the Nextcloud.
 
         :param request: The `Starlette request <https://www.starlette.io/requests/>`_
@@ -535,7 +535,7 @@ async def unregister_talk_bot(self, callback_url: str) -> bool:
             return False
         return True
 
-    def request_sign_check(self, request: FastAPIRequest) -> bool:
+    def request_sign_check(self, request: HTTPConnection) -> bool:
         """Verifies the signature and validity of an incoming request from the Nextcloud.
 
         :param request: The `Starlette request <https://www.starlette.io/requests/>`_

tests/_install_async.py

@@ -1,7 +1,6 @@
 from contextlib import asynccontextmanager
-from typing import Annotated
 
-from fastapi import Depends, FastAPI
+from fastapi import FastAPI
 from fastapi.responses import JSONResponse
 
 from nc_py_api import AsyncNextcloudApp, ex_app
@@ -14,12 +13,12 @@ async def lifespan(_app: FastAPI):
 
 
 APP = FastAPI(lifespan=lifespan)
+APP.add_middleware(ex_app.AppAPIAuthMiddleware)
 
 
 @APP.put("/sec_check")
 async def sec_check(
     value: int,
-    _nc: Annotated[AsyncNextcloudApp, Depends(ex_app.anc_app)],
 ):
     print(value, flush=True)
     return JSONResponse(content={"error": ""}, status_code=200)