Prepare monorepo

backend/app/security/jwt_bearer.py

@@ -0,0 +1,47 @@
+from typing import Optional
+from fastapi import Request, HTTPException
+from fastapi.params import Depends
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.database import get_db
+from app.services import token_service
+from app.util.errors import InsufficientPermissionsError, InvalidTokenAudienceError
+from app.schemas.auth_token import AccessToken
+
+
+class JWTBearer(HTTPBearer):
+    __required_roles: list[str] | None
+
+    def __init__(
+        self, required_roles: Optional[list[str]] = None, auto_error: bool = True
+    ):
+        self.__required_roles = required_roles
+        super(JWTBearer, self).__init__(auto_error=auto_error)
+
+    async def __call__(
+        self, request: Request, db: AsyncSession = Depends(get_db)
+    ) -> AccessToken:
+        credentials: HTTPAuthorizationCredentials | None = await super(
+            JWTBearer, self
+        ).__call__(request)
+        if credentials:
+            if not credentials.scheme == "Bearer":
+                raise HTTPException(
+                    status_code=403, detail="authentication_scheme_invalid"
+                )
+            try:
+                token = await token_service.verify_access_token(
+                    credentials.credentials, self.__required_roles
+                )
+                if not token:
+                    raise HTTPException(
+                        status_code=403, detail="token_invalid_or_expired"
+                    )
+                return token
+            except InsufficientPermissionsError:
+                raise HTTPException(status_code=403, detail="insufficient_permissions")
+            except InvalidTokenAudienceError:
+                raise HTTPException(status_code=403, detail="invalid_token_audience")
+        else:
+            raise HTTPException(status_code=403, detail="authorization_code_invalid")

backend/app/security/websocket_auth_backend.py

@@ -0,0 +1,33 @@
+import base64
+import binascii
+from starlette.authentication import (
+    AuthCredentials, AuthenticationBackend, AuthenticationError
+)
+
+from app.database import SessionLocal
+from app.models.chargepoint import ChargePoint
+
+class BasicAuthBackend(AuthenticationBackend):
+    async def authenticate(self, conn):
+        if "Authorization" not in conn.headers:
+            raise AuthenticationError('No Authorization header provided')
+
+        auth = conn.headers["Authorization"]
+        try:
+            scheme, credentials = auth.split()
+            if scheme.lower() != 'basic':
+                raise AuthenticationError('Invalid authorization scheme')
+            decoded = base64.b64decode(credentials).decode("ascii")
+        except (ValueError, UnicodeDecodeError, binascii.Error) as exc:
+            raise AuthenticationError('Invalid basic auth credentials')
+
+        username, _, password = decoded.partition(":")
+        
+        with SessionLocal() as db:
+            chargepoint = db.query(ChargePoint).filter(ChargePoint.identity == username).first()
+            if chargepoint is None:
+                raise AuthenticationError('Invalid basic auth credentials')
+            if chargepoint.password != password:
+                raise AuthenticationError('Invalid basic auth credentials')
+            
+        return AuthCredentials(["authenticated"]), chargepoint