Merge pull request #12 from ag-gipp/issue-8

Issue 8

.flake8

@@ -3,5 +3,5 @@ exclude = .venv, venv/
 docstring-convention = google
 max-line-length = 100
 
-ignore = D104, D205, D415
-per-file-ignores = tests/*: D1
+ignore = D104, D205, D415, W503
+per-file-ignores = tests/*: D1

.gitignore

@@ -132,4 +132,7 @@ dmypy.json
 .vercel
 .DS_Store
 docs/oas.json
-poetry.lock
+poetry.lock
+
+# vim specific swap files
+*.swp

.vscode/settings.json

@@ -3,10 +3,11 @@
   "python.linting.lintOnSave": true,
   "python.linting.mypyEnabled": true,
   "python.linting.flake8Enabled": true,
+  "mypy.runUsingActiveInterpreter": true,
   "python.formatting.provider": "black",
   "python.formatting.blackArgs": [
     "--line-length",
-    "99",
+    "100",
     "--experimental-string-processing"
   ],
   "editor.formatOnSave": true,

mypy.ini

@@ -2,4 +2,5 @@
 python_version = 3.8
 warn_return_any = True
 warn_unused_configs = True
-disallow_untyped_defs = True
+disallow_untyped_defs = True
+exclude = .venv, venv/

nlp_land_prediction_endpoint/app.py

@@ -1,12 +1,35 @@
 """This module implements the main app."""
+import os
+
+if not os.path.exists("./.env"):
+    print("======== No .env file found ========")
+    print("= Copy the contents of sample.env  =")
+    print("= to .env in the root directory    =")
+    print("= and change the contents!         =")
+    print("====================================")
+    os.environ["AUTH_BACKEND_VERSION"] = "v0"
+    os.environ["AUTH_BACKEND_URL"] = "http://127.0.0.1/api/{version}"
+    os.environ["AUTH_BACKEND_LOGIN_ROUTE"] = "/auth/login/service"
+    os.environ["AUTH_TOKEN_ROUTE"] = "/auth/service"
+    os.environ["JWT_SECRET"] = "super_secret_secret"
+    os.environ["JWT_TOKEN_EXPIRATION_MINUTES"] = "30"
+    os.environ["JWT_SIGN_ALG"] = "HS256"
+
+
+from decouple import config  # type: ignore
 from fastapi import FastAPI
 
 import nlp_land_prediction_endpoint
+from nlp_land_prediction_endpoint.routes.route_auth import router as AuthRouter
 from nlp_land_prediction_endpoint.routes.route_status import router as StatusRouter
 from nlp_land_prediction_endpoint.routes.route_topic import router as TopicRouter
+from nlp_land_prediction_endpoint.utils.version_getter import get_backend_version
 
 app = FastAPI(title="NLP-Land-prediction-endpoint", docs_url="/api/docs", redoc_url="/api/redoc")
 
+if "{version}" in config("AUTH_BACKEND_URL"):
+    get_backend_version()
+
 
 # app.add_event_handler("startup", connect_to_third_party_services)
 # app.add_event_handler("shutdown", close_third_party_services)
@@ -16,8 +39,15 @@
     tags=["Status"],
     prefix=f"/api/v{nlp_land_prediction_endpoint.__version__.split('.')[0]}/status",
 )
+
 app.include_router(
     TopicRouter,
     tags=["Topics"],
     prefix=f"/api/v{nlp_land_prediction_endpoint.__version__.split('.')[0]}/topics",
 )
+
+app.include_router(
+    AuthRouter,
+    tags=["Auth"],
+    prefix=f"/api/v{nlp_land_prediction_endpoint.__version__.split('.')[0]}/auth",
+)

nlp_land_prediction_endpoint/middleware/auth.py

@@ -0,0 +1,118 @@
+"""Middlware that allows for protection of endoints using JWTs"""
+from datetime import datetime, timedelta
+from typing import Optional
+
+import jwt
+import pydantic
+import requests  # type: ignore
+from decouple import config  # type: ignore
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+
+from nlp_land_prediction_endpoint.models.model_token_data import TokenData
+from nlp_land_prediction_endpoint.models.model_user import UserModel
+from nlp_land_prediction_endpoint.models.model_user_login import UserLoginModel
+
+token_url = config("AUTH_TOKEN_ROUTE")
+jwt_scheme = OAuth2PasswordBearer(tokenUrl=token_url)
+
+
+def encode_token(data: dict) -> str:
+    """Encodes supplied data into an JWT
+    Arguments:
+        data (dict): Dictionary containing some data (e.g., a UserModel dict)
+
+    Returns:
+        str: a valid JWT token
+    """
+    SECRET = config("JWT_SECRET")
+    ALG = config("JWT_SIGN_ALG")
+    return jwt.encode(data, SECRET, ALG)
+
+
+def decode_token(token: str) -> TokenData:
+    """Decodes a supplied JWT into a TokenData model
+    (decoding exception should be catched on function call)
+
+    Arguments:
+        token (str): a (possibly invalid) JWT token
+
+    Returns:
+        TokenData: a TokenData model representing the decoded JWT token
+    """
+    SECRET = config("JWT_SECRET")
+    ALG = config("JWT_SIGN_ALG")
+    return TokenData(**jwt.decode(token, SECRET, [ALG]))
+
+
+def create_token(user: UserModel, expires_delta: timedelta = None) -> str:
+    """Creates a JWT given a user as TokenData.
+    This will use the JWT_SECRET and JWT_SIGN_ALG as defined in the .env variable.
+
+    Arguments:
+        user (TokenData): TokenData object containing at minimum the email
+        expires_delta (timedelta): time offset from NOW when the token will expire
+
+    Returns:
+        str: a valid JWT as a string
+    """
+    data = user.dict().copy()
+    if expires_delta:
+        expires = datetime.utcnow() + expires_delta
+    else:
+        expires = datetime.utcnow() + timedelta(minutes=30)
+    data.update({"sub": user.email})
+    data.update({"exp": expires})
+    token = encode_token(data)
+    return token
+
+
+def authenticate_user(user: UserLoginModel) -> Optional[UserModel]:
+    """Checks whether the supplied UserModel contains valid
+    credentials. This is done by going through the authorization
+    endpoint specified in AUTH_LOGIN_ROUTE at the host AUTH_LOGIN_PROVIDER.
+
+    Arguments:
+        user (UserModel): a user model to authenticate
+
+    Returns:
+        Optional[UserModel]: If the authentication was successful a UserModel object;
+        None otherwise
+    """
+    login_provider = config("AUTH_BACKEND_URL")
+    login_route = config("AUTH_BACKEND_LOGIN_ROUTE")
+    try:
+        r = requests.post(
+            f"{login_provider}{login_route}",
+            data=user.dict(),
+            headers={"content-type": "application/json"},
+        )
+        if r.status_code == status.HTTP_200_OK:
+            return UserModel(**r.json())
+        else:
+            return None
+    except requests.RequestException:
+        return None
+
+
+async def get_current_user(token: str = Depends(jwt_scheme)) -> UserModel:
+    """Returns the current user given a valid JWT
+
+    Arguments:
+        token (str): a bearer token taken from the "Authorization" header
+
+    Returns:
+        UserModel: If the token is valid a UserModel with at least an email;
+        None otherwise
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        decoded_token = decode_token(token)
+        user = UserModel(**decoded_token.dict())
+    except (jwt.exceptions.InvalidTokenError, pydantic.ValidationError):
+        raise credentials_exception
+    return user

nlp_land_prediction_endpoint/models/model_token.py

@@ -0,0 +1,14 @@
+"""Model used for JWT-responses"""
+from pydantic import BaseModel, Field
+
+
+class TokenModel(BaseModel):
+    """Model used for JWT-responses
+
+    Attributes:
+        acces_token (str): the actual token
+        token_type (str): this will always be bearer
+    """
+
+    access_token: str = Field(...)
+    token_type: str = Field(...)

nlp_land_prediction_endpoint/models/model_token_data.py

@@ -0,0 +1,25 @@
+"""Model used for JWT-token data"""
+from typing import Optional
+
+from pydantic import Field
+
+from nlp_land_prediction_endpoint.models.model_user import UserModel
+
+
+class TokenData(UserModel):
+    """Model used for Token data (the payload of the token)
+    This contains a subset of attributs from the UserModel
+
+    Attributes:
+        email (str): email of the user
+        fullname (str): fullname of the user
+        isAdmin (Optional[bool]): flag indicating whether it is a admin
+        isActive (Optional[bool]): flag indicating whether the user is still active
+
+        sub (str): subject of the JWT (email)
+        exp (str): expiration date of the JWT
+    """
+
+    # JWT specific attributes
+    sub: Optional[str] = Field(default=None)  # Subject (unique identifier)
+    exp: Optional[str] = Field(default=None)  # Expiration (expiration date)

nlp_land_prediction_endpoint/models/model_user.py

@@ -0,0 +1,42 @@
+"""Model used for defining a user; Should be equivilent to the
+definition in the NLP-Land-Backend
+"""
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class UserModel(BaseModel):
+    """The model for a user. Should be identical to the NLP-Land-backend
+
+    Arguments:
+        BaseModel(Any): Base class of FastAPI modesl.
+
+    Attributes:
+        email (str): email of the user
+        password (str): password of the user
+        fullname (Optional[str]): fullname of the user
+        isAdmin (Optional[bool]): flag indicating whether it is a admin
+        isActive (Optional[bool]): flag indicating whether the user is still active
+    """
+
+    email: str = Field(...)
+    fullname: Optional[str] = Field()
+    isAdmin: Optional[bool] = Field()
+    isActive: Optional[bool] = Field()
+
+    # XXX-TN change isAdmin to groups
+    # TODO-TN isActive is currently getting ignored
+
+    class Config:
+        """Configuration for UserModel"""
+
+        schema_extra = {
+            "example": {
+                "email": "admin@nlp.de",
+                "fullname=": "admin",
+                "token": None,
+                "isAdmin": True,
+                "isActive": True,
+            }
+        }

nlp_land_prediction_endpoint/models/model_user_login.py

@@ -0,0 +1,30 @@
+"""Model used for defining a user; Should be equivilent to the
+definition in the NLP-Land-Backend
+"""
+from pydantic import BaseModel, Field
+
+
+class UserLoginModel(BaseModel):
+    """The model for a user login. This model contains only
+    the fields that are submitted on a login
+
+    Arguments:
+        BaseModel(Any): Base class of FastAPI modesl.
+
+    Attributes:
+        email (str): email of the user
+        password (str): password of the user
+    """
+
+    email: str = Field(...)
+    password: str = Field(...)
+
+    class Config:
+        """Configuration for UserModel"""
+
+        schema_extra = {
+            "example": {
+                "email": "admin@nlp.de",
+                "password": "12345",
+            }
+        }

nlp_land_prediction_endpoint/routes/route_auth.py

@@ -0,0 +1,56 @@
+"""This module implements the authentication endpoint"""
+from datetime import timedelta
+
+from decouple import config  # type: ignore
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from nlp_land_prediction_endpoint.middleware.auth import (
+    authenticate_user,
+    create_token,
+    get_current_user,
+)
+from nlp_land_prediction_endpoint.models.model_token import TokenModel
+from nlp_land_prediction_endpoint.models.model_token_data import TokenData
+from nlp_land_prediction_endpoint.models.model_user import UserModel
+from nlp_land_prediction_endpoint.models.model_user_login import UserLoginModel
+
+router = APIRouter()
+
+TIME_DELTA = config("JWT_TOKEN_EXPIRATION_MINUTES", cast=int)
+
+
+@router.post("/login", response_description="Trigger login procedure", response_model=TokenModel)
+async def login(user: UserLoginModel) -> TokenModel:
+    """Login routine
+
+    Arguments:
+        user (UserModel): user credentials at least (email, password)
+
+    Returns:
+        TokenModel: a JWT given a valid user from the NLP-Land-Backend
+    """
+    auth_user = authenticate_user(user)
+    if not auth_user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    token_expiration = timedelta(minutes=TIME_DELTA)
+    token = create_token(auth_user, token_expiration)
+    return TokenModel(access_token=token, token_type="bearer")
+
+
+@router.post("/refresh", response_description="Trigger login procedure", response_model=TokenModel)
+async def refresh(user: UserModel = Depends(get_current_user)) -> TokenModel:
+    """Generates a new token without the need of logging in again
+
+    Arguments:
+        user(UserModel): a user depending on the supplied token
+
+    Returns:
+        TokenModel: a JWT given a valid user from the NLP-Land-Backend
+    """
+    token_expiration = timedelta(minutes=TIME_DELTA)
+    token = create_token(TokenData(**user.dict()), token_expiration)
+    return TokenModel(access_token=token, token_type="bearer")