added routes and oauth

.gitignore

@@ -127,3 +127,6 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# google cert
+service-key.json

README.md

@@ -1,2 +1,23 @@
 # Certify
 IITM-POD Certificate Verification System Portal
+
+
+# Setup
+Require **Python 3.8+**. Create a virtualenv,
+
+```bash
+python -m virtualenv env
+cd env
+```
+
+Install [Poetry](https://python-poetry.org/) (`pip install poetry`), and run poetry to install dependencies.
+
+```bash
+poetry install
+```
+
+Deploy the server using `uvicorn`
+
+```bash
+uvicorn main:app
+```

certify/constants/jwt.py

@@ -0,0 +1,5 @@
+from enum import Enum
+
+class JWTTokenType(Enum):
+    HS256 = "HS256"
+    RS256 = "RS256"

certify/constants/scope.py

@@ -0,0 +1,14 @@
+from enum import Enum
+
+
+class Scope(Enum):
+    Master = "master"
+
+    ViewUserCert = "cert:view"
+    ViewAllUserCert = "cert:view:all"
+    CreateCert = "cert:create"
+    EditCert = "cert:edit"
+
+    def __str__(self):
+        return str(self.value)
+

certify/core/config.py

@@ -0,0 +1,35 @@
+import logging
+import sys
+from typing import List
+
+from loguru import logger
+from starlette.config import Config
+from starlette.datastructures import CommaSeparatedStrings, Secret, URL
+
+from certify.core.logging import InterceptHandler 
+from certify.constants.jwt import JWTTokenType
+
+config = Config(".env")
+
+DEBUG = config("DEBUG", cast=bool, default=False)
+SECRET_KEY: Secret = config(
+    "SECRET_KEY", cast=Secret, default="5df9db467ed2c905bcc1")
+ALLOWED_HOSTS: List[str] = config(
+    "ALLOWED_HOSTS", cast=CommaSeparatedStrings, default=[]
+)
+LOGGING_LEVEL = (
+    logging.DEBUG
+    if DEBUG
+    else config("LOGGING_LEVEL", cast=lambda x: getattr(logging, x), default="INFO")
+)
+DEFAULT_TOKEN_EXPIRE = config(
+    "DEFAULT_TOKEN_EXPIRE", cast=int, default=15 * 60)
+JWT_ALGORITHM = config("DEFAULT_TOKEN_EXPIRE",
+                       cast=JWTTokenType, default="HS256")
+
+logging.getLogger().handlers = [InterceptHandler()]
+LOGGERS = ("uvicorn.asgi", "uvicorn.access")
+for logger_name in LOGGERS:
+    logging_logger = logging.getLogger(logger_name)
+    logging_logger.setLevel(logging.INFO)
+    logging_logger.handlers = [InterceptHandler(level=LOGGING_LEVEL)]

certify/core/error/http_error.py

@@ -0,0 +1,31 @@
+from fastapi import HTTPException
+from fastapi.encoders import jsonable_encoder
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from pydantic import BaseModel
+
+
+async def http_error_handler(_: Request, exc: HTTPException) -> JSONResponse:
+    """Intercept any/all HTTPExceptions from FastAPI, and return a JSON response.
+    Args:
+        _ (Request)
+        exc (HTTPException)
+    Returns:
+        JSONResponse
+    """
+    if isinstance(exc.detail, BaseModel):
+        return JSONResponse(
+            {
+                "has_error": True,
+                "success": False,
+                "data": None,
+                "error": jsonable_encoder(exc.detail),
+            },
+            status_code=exc.status_code,
+        )
+
+    return JSONResponse(
+        {"has_error": True, "success": False,
+            "data": None, "error": str(exc.detail)},
+        status_code=exc.status_code,
+    )

certify/core/error/validation_error.py

@@ -0,0 +1,35 @@
+from typing import Union
+
+from fastapi.exceptions import RequestValidationError
+from fastapi.openapi.constants import REF_PREFIX
+from fastapi.openapi.utils import validation_error_response_definition
+from pydantic import ValidationError
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
+
+
+async def http422_error_handler(
+    _: Request,
+    exc: Union[RequestValidationError, ValidationError],
+) -> JSONResponse:
+    """Intercept any/all ValidationError and return a JSON response.
+    Args:
+        _ (Request)
+        exc (Union[RequestValidationError, ValidationError])
+    Returns:
+        JSONResponse
+    """
+    return JSONResponse(
+        {"has_error": True, "success": False, "data": None, "error": exc.errors()},
+        status_code=HTTP_422_UNPROCESSABLE_ENTITY,
+    )
+
+
+validation_error_response_definition["properties"] = {
+    "errors": {
+        "title": "Errors",
+        "type": "array",
+        "items": {"$ref": "{0}ValidationError".format(REF_PREFIX)},
+    },
+}

certify/core/events.py

@@ -0,0 +1,33 @@
+from typing import Callable
+
+from fastapi import FastAPI
+from loguru import logger
+
+
+def create_start_app_handler(app: FastAPI) -> Callable:
+    """FastAPI start app event
+    Args:
+        app (FastAPI)
+    Returns:
+        Callable
+    """
+
+    async def start_app() -> None:
+        logger.info("Connecting to Firebase database")
+
+    return start_app
+
+
+def create_stop_app_handler(app: FastAPI) -> Callable:
+    """FastAPI shutdown event
+    Args:
+        app (FastAPI)
+    Returns:
+        Callable
+    """
+
+    @logger.catch
+    async def stop_app() -> None:
+        logger.info("Disconnecting Firebase database")
+
+    return stop_app

certify/core/logging.py

@@ -0,0 +1,25 @@
+import logging
+from types import FrameType
+from typing import cast
+from loguru import logger
+
+
+class InterceptHandler(logging.Handler):
+    def emit(self, record: logging.LogRecord) -> None:  # pragma: no cover
+        # Get corresponding Loguru level if it exists
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = str(record.levelno)
+
+        # Find caller from where originated the logged message
+        frame, depth = logging.currentframe(), 2
+        while frame.f_code.co_filename == logging.__file__:
+            frame = cast(FrameType, frame.f_back)
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info, colors=True).log(
+            level,
+            record.getMessage(),
+        )
+

certify/core/middleware/jwt.py

@@ -0,0 +1,39 @@
+from fastapi.security.utils import get_authorization_scheme_param
+from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.requests import Request
+from typing import Dict, Optional, Tuple
+from jose import jwt
+
+from certify.core.config import SECRET_KEY, JWT_ALGORITHM
+
+
+class OAuthMiddleware:
+
+    def __init__(self, app: ASGIApp):
+        self._app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        if scope["type"] in ("http", "websocket"):
+            req: Request = Request(scope, receive)
+            token: str = await self.retrieve_oauth_token(req)
+
+            try:
+                payload = jwt.decode(
+                    token, str(SECRET_KEY), algorithms=[JWT_ALGORITHM.value]
+                )
+
+                scope["oauth"] = payload
+
+            except:
+                pass
+
+        await self._app(scope, receive, send)
+
+    async def retrieve_oauth_token(self, request: Request) -> Optional[str]:
+        authorization: str = request.headers.get("Authorization")
+        scheme, param = get_authorization_scheme_param(authorization)
+        if not authorization or scheme.lower() != "bearer":
+            return None
+
+        return param
+

certify/models/response.py

@@ -0,0 +1,12 @@
+from pydantic.generics import GenericModel
+from typing import TypeVar, Optional, Generic, Any
+
+T = TypeVar("T")
+
+
+class Response(GenericModel, Generic[T]):
+    data: Optional[T] = None
+    error: Any = None
+
+    success: bool
+    hasError: bool = False

certify/models/token.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel
+from typing import Optional
+
+from certify.models.response import Response
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+    session_key: Optional[str]
+
+
+class TokenError(BaseModel):
+    error_type: str
+    error_code: int
+    error_description: str
+
+
+class TokenResponse(Response[Token]):
+    error: TokenError = None
+    has_error: bool = False

certify/routes/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from certify.routes import api
+
+router = APIRouter()
+router.include_router(api.router, prefix="/api", tags=["CertAPI"])

certify/routes/api/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from certify.routes.api import v1
+
+router = APIRouter()
+#router.include_router(v1.router, prefix="/v1", tags=["API v1"])