Config: Add the read_only setting

README.md

@@ -34,6 +34,18 @@ uvicorn aiida_restapi:app
 uvicorn aiida_restapi:app --reload
 ```
 
+By default all endpoints of the REST API are available.
+The API can be made *read-only* by setting the `read_only` configuration settings to `True`.
+This can either be done by setting the environment variable:
+```bash
+export READ_ONLY=True
+```
+or by adding the following to the `.env` file:
+```ini
+read_only=true
+```
+When the API is read-only, all `DELETE`, `PATCH`, `POST` and `PUT` requests will result in a `405 - Method Not Allowed` response.
+
 ## Examples
 
 See the [examples](https://github.com/aiidateam/aiida-restapi/tree/master/examples) directory.

aiida_restapi/config.py

@@ -1,11 +1,49 @@
 """Configuration of API"""
 
+from functools import lru_cache
+
+from pydantic_settings import BaseSettings
+
 # to get a string like this run:
 # openssl rand -hex 32
 SECRET_KEY = '09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7'
 ALGORITHM = 'HS256'
 ACCESS_TOKEN_EXPIRE_MINUTES = 30
 
+
+class Settings(BaseSettings):
+    """Configuration settings for the application."""
+
+    # pylint: disable=too-few-public-methods
+
+    class Config:
+        """Config settings."""
+
+        env_file = '.env'
+        env_file_encoding = 'utf-8'
+
+    secret_key: str = '09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7'
+    """The secret key used to create access tokens."""
+
+    secret_key_algoritm: str = 'HS256'
+    """The algorithm used to create access tokens."""
+
+    access_token_expire_minutes: int = 30
+    """The number of minutes an access token remains valid."""
+
+    read_only: bool = False
+    """Whether the instance is read-only. If set to ``True`` all DELETE, PATCH, POST and PUT methods will raise 405."""
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    """Return the configuration settings for the application.
+
+    This function is cached and should be used preferentially over constructing ``Settings`` directly.
+    """
+    return Settings()
+
+
 fake_users_db = {
     '[email protected]': {
         'pk': 23,

aiida_restapi/main.py

@@ -5,7 +5,12 @@
 from aiida_restapi.graphql import main
 from aiida_restapi.routers import auth, computers, daemon, groups, nodes, process, users
 
+from .middleware import protected_methods_middleware
+
 app = FastAPI()
+
+app.middleware('http')(protected_methods_middleware)
+
 app.include_router(auth.router)
 app.include_router(computers.router)
 app.include_router(daemon.router)

aiida_restapi/middleware.py

@@ -0,0 +1,26 @@
+"""Module with middleware."""
+
+from typing import Callable
+
+from fastapi import Request, Response
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
+
+from .config import Settings, get_settings
+
+
+async def protected_methods_middleware(request: Request, call_next: Callable[[Request], Response]) -> Response:
+    """Middleware that will return a 405 if the instance is read only and the request method is mutating.
+
+    Mutating request methods are `DELETE`, `PATCH`, `POST`, `PUT`.
+    """
+    settings: Settings = get_settings()
+
+    if settings.read_only and request.method in {'DELETE', 'PATCH', 'POST', 'PUT'}:
+        return JSONResponse(
+            status_code=405,
+            content=jsonable_encoder({'reason': 'This instance is read-only.'}),
+            media_type='application/json',
+        )
+
+    return await call_next(request)

aiida_restapi/routers/auth.py

@@ -10,9 +10,10 @@
 from passlib.context import CryptContext
 from pydantic import BaseModel
 
-from aiida_restapi import config
 from aiida_restapi.models import User
 
+from ..config import Settings, fake_users_db, get_settings
+
 
 class Token(BaseModel):
     access_token: str
@@ -62,32 +63,32 @@ def authenticate_user(fake_db: dict, email: str, password: str) -> Optional[User
     return user
 
 
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+def create_access_token(settings: Settings, data: dict, expires_delta: Optional[timedelta] = None) -> str:
     to_encode = data.copy()
     if expires_delta:
         expire = datetime.utcnow() + expires_delta
     else:
         expire = datetime.utcnow() + timedelta(minutes=15)
     to_encode.update({'exp': expire})
-    encoded_jwt = jwt.encode(to_encode, config.SECRET_KEY, algorithm=config.ALGORITHM)
+    encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.secret_key_algoritm)
     return encoded_jwt
 
 
-async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
+async def get_current_user(token: str = Depends(oauth2_scheme), settings: Settings = Depends(get_settings)) -> User:
     credentials_exception = HTTPException(
         status_code=status.HTTP_401_UNAUTHORIZED,
         detail='Could not validate credentials',
         headers={'WWW-Authenticate': 'Bearer'},
     )
     try:
-        payload = jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
+        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.secret_key_algoritm])
         email: str = payload.get('sub')
         if email is None:
             raise credentials_exception
         token_data = TokenData(email=email)
     except JWTError:
         raise credentials_exception  # pylint: disable=raise-missing-from
-    user = get_user(config.fake_users_db, email=token_data.email)
+    user = get_user(fake_users_db, email=token_data.email)
     if user is None:
         raise credentials_exception
     return user
@@ -104,16 +105,17 @@ async def get_current_active_user(
 @router.post('/token', response_model=Token)
 async def login_for_access_token(
     form_data: OAuth2PasswordRequestForm = Depends(),
+    settings: Settings = Depends(get_settings),
 ) -> Dict[str, Any]:
-    user = authenticate_user(config.fake_users_db, form_data.username, form_data.password)
+    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
     if not user:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail='Incorrect email or password',
             headers={'WWW-Authenticate': 'Bearer'},
         )
-    access_token_expires = timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = create_access_token(data={'sub': user.email}, expires_delta=access_token_expires)
+    access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
+    access_token = create_access_token(settings, data={'sub': user.email}, expires_delta=access_token_expires)
     return {'access_token': access_token, 'token_type': 'bearer'}
 
 

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
   'fastapi~=0.115.5',
   'uvicorn[standard]~=0.19.0',
   'pydantic~=2.0',
+  'pydantic-settings',
   'starlette-graphene3~=0.6.0',
   'graphene~=3.0',
   'python-dateutil~=2.0',