WiP first version

Author: mrtolkien

.gitignore

@@ -127,3 +127,9 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# IDE files
+.idea/
+
+# DB files
+*.db

Dockerfile

@@ -0,0 +1,4 @@
+FROM tiangolo/uvicorn-gunicorn-fastapi:latest
+
+# Installing dev depedencies
+RUN pip install pytest python-dotenv

app/main.py

@@ -0,0 +1,18 @@
+from fastapi import FastAPI, Depends
+import fastapi_simple_security
+
+
+app = FastAPI(title="FastAPI simple security", version="test")
+
+
+@app.get("/open")
+async def root():
+    return {"message": "This is an unsecured endpoint"}
+
+
+@app.get("/secure", dependencies=[Depends(fastapi_simple_security.api_key_security)])
+async def root():
+    return {"message": "This is a secured endpoint"}
+
+
+app.include_router(fastapi_simple_security.api_key_router, prefix="/auth", tags=["_auth"])

docker-compose.yml

@@ -0,0 +1,21 @@
+version: "3.8"
+
+services:
+  api:
+    image: fastapi_simple_security
+    container_name: fastapi_simple_security
+    restart: always
+    build:
+      context: .
+      dockerfile: Dockerfile
+    ports:
+      - target: 80
+        published: 8080
+    volumes:
+      - type: bind
+        source: ./app
+        target: /app
+      - type: bind
+        source: ./fastapi_simple_security
+        target: /app/fastapi_simple_security
+    command: /start-reload.sh

fastapi_simple_security/__init__.py

@@ -0,0 +1,2 @@
+from fastapi_simple_security.router import api_key_router
+from fastapi_simple_security.security_api_key import api_key_security

fastapi_simple_security/_security_secret.py

@@ -0,0 +1,48 @@
+import os
+import uuid
+import warnings
+
+from fastapi import Security
+from fastapi.security import APIKeyHeader
+from starlette.exceptions import HTTPException
+from starlette.status import HTTP_403_FORBIDDEN
+
+try:
+    SECRET = os.environ["FASTAPI_SIMPLE_SECURITY_SECRET"]
+except KeyError:
+    SECRET = str(uuid.uuid4())
+
+    warnings.warn(
+        f"ENVIRONMENT VARIABLE 'FASTAPI_SIMPLE_SECURITY_SECRET' NOT FOUND\n"
+        f"\tGenerated a single-use secret key for this session:\n"
+        f"\t{SECRET=}"
+    )
+
+SECRET_KEY_NAME = "secret_key"
+
+secret_header = APIKeyHeader(name=SECRET_KEY_NAME, scheme_name="Secret header", auto_error=False)
+
+
+async def secret_based_security(header_param: str = Security(secret_header)):
+    """
+    Args:
+        header_param: parsed header field secret_header
+
+    Returns:
+        True if the authentication was successful
+
+    Raises:
+        HTTPException if the authentication failed
+    """
+
+    if header_param == SECRET:
+        return True
+    if not header_param:
+        error = "secret_key must be passed as a header field"
+    else:
+        error = (
+            "Wrong secret key. If not set through environment variable 'FASTAPI_SIMPLE_SECURITY_SECRET', it was "
+            "generated automatically at startup and appears in the server logs."
+        )
+
+    raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail=error)

fastapi_simple_security/_sqlite_access.py

@@ -0,0 +1,2 @@
+class SQLiteAccess:
+    pass

fastapi_simple_security/router.py

@@ -0,0 +1,43 @@
+from fastapi import APIRouter, Depends
+import uuid
+
+from fastapi_simple_security._security_secret import secret_based_security
+
+api_key_router = APIRouter()
+
+
+# TODO Add an environment variable to decide if it’s in docs or not
+
+
+@api_key_router.get("/new", dependencies=[Depends(secret_based_security)])
+def get_new_api_key() -> str:
+    """
+    Returns:
+        api_key: a newly generated API key
+    """
+    # TODO Add API key to sqlite db with creation date
+
+    return str(uuid.uuid4())
+
+
+@api_key_router.get("/revoke", dependencies=[Depends(secret_based_security)])
+def revoke_api_key(api_key: str):
+    """
+    Revokes the usage of the given API key. Does not remove its logs.
+
+    Args:
+        api_key: the api_key to revoke
+
+    Return:
+        Status 200 on successful revocation
+
+    Raises:
+        Status 404 if the API key was not in the database
+    """
+    return api_key
+
+
+@api_key_router.get("/logs", dependencies=[Depends(secret_based_security)])
+def get_api_key_usage_logs():
+    # TODO Get usage logs per api key
+    pass

fastapi_simple_security/security_api_key.py

@@ -0,0 +1,21 @@
+from fastapi import Security
+from fastapi.security import APIKeyQuery, APIKeyHeader
+from starlette.exceptions import HTTPException
+from starlette.status import HTTP_403_FORBIDDEN
+
+API_KEY_NAME = "access_token"
+
+api_key_query = APIKeyQuery(name=API_KEY_NAME, scheme_name="API key query", auto_error=False)
+api_key_header = APIKeyHeader(name=API_KEY_NAME, scheme_name="API key header", auto_error=False)
+
+
+async def api_key_security(
+    query_param: str = Security(api_key_query), header_param: str = Security(api_key_header),
+):
+    # TODO API key testing logic here
+    if query_param == "12345":
+        return query_param
+    elif header_param == "12345":
+        return header_param
+    else:
+        raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail="Could not validate credentials")