Fix email security problem, switch to host located in settings (#20)

* Merge tests in fix brach 

* Fixtures, "forgot password" tests init, tests refactoring

* email tests

* fix path, fix security

* deploy fix

* add check tokens test

Author: grigoriev-semyon

.github/workflows/build_and_publish.yml

@@ -87,6 +87,7 @@ jobs:
             --env EMAIL_PASS='${{ secrets.EMAIL_PASS }}' \
             --env SMTP_HOST='smtp.gmail.com' \
             --env SMTP_PORT='587' \
+            --env HOST='${{ secrets.HOST }}' \
             --name ${{ env.CONTAITER_NAME }} \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:test
 
@@ -131,5 +132,6 @@ jobs:
             --env ENABLED_AUTH_METHODS='${{ secrets.ENABLED_AUTH_METHODS }}' \
             --env SMTP_HOST='smtp.gmail.com' \
             --env SMTP_PORT='587' \
+            --env HOST='${{ secrets.HOST }}' \
             --name ${{ env.CONTAITER_NAME }} \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest

auth_backend/auth_plugins/auth_method.py

@@ -6,14 +6,13 @@
 from fastapi import APIRouter
 
 from datetime import datetime
-from auth_backend.base import Base
+from auth_backend.base import Base, Token
 
 
-class Session(Base):
+class Session(Token):
     expires: datetime
     id: int
     user_id: int
-    token: str
 
 
 AUTH_METHODS: dict[str, type[AuthMethodMeta]] = {}

auth_backend/auth_plugins/email.py

@@ -1,19 +1,20 @@
 import hashlib
 import random
 import string
+
+from fastapi import Request
 from fastapi_sqlalchemy import db
-from fastapi import Request, HTTPException
 from pydantic import validator, constr
+from sqlalchemy import func
 from starlette.responses import JSONResponse
 
+from auth_backend.base import Base, ResponseModel
 from auth_backend.exceptions import AlreadyExists, AuthFailed, ObjectNotFound, SessionExpired
 from auth_backend.models.db import AuthMethod
 from auth_backend.models.db import UserSession, User
 from auth_backend.settings import get_settings
 from auth_backend.utils.smtp import send_confirmation_email
 from .auth_method import AuthMethodMeta, Session
-from auth_backend.base import Base, ResponseModel
-from sqlalchemy import func
 
 settings = get_settings()
 
@@ -24,8 +25,12 @@ class EmailLogin(Base):
 
     @validator('email')
     def check_email(cls, v):
+        restricted: set[str] = {'"', '#', '&', "'", '(', ')', '*', ',', '/', ';', '<', '>', '?',
+                      '[', '\\', ']', '^', '`', '{', '|', '}', '~', '\n', '\r'}
         if "@" not in v:
             raise ValueError()
+        if set(v) & restricted:
+            raise ValueError()
         return v
 
 
@@ -51,12 +56,12 @@ def __init__(self):
     async def login(user_inp: EmailLogin) -> Session:
         query = (
             db.session.query(AuthMethod)
-            .filter(
+                .filter(
                 func.lower(AuthMethod.value) == user_inp.email.lower(),
                 AuthMethod.param == "email",
                 AuthMethod.auth_method == Email.get_name(),
             )
-            .one_or_none()
+                .one_or_none()
         )
         if not query:
             raise AuthFailed(error="Incorrect login or password")
@@ -66,7 +71,7 @@ async def login(user_inp: EmailLogin) -> Session:
                 error="Registration wasn't completed. Try to registrate again and do not forget to approve your email"
             )
         if secrets.get("email").lower() != user_inp.email.lower() or not Email.validate_password(
-            user_inp.password, secrets.get("hashed_password"), secrets.get("salt")
+                user_inp.password, secrets.get("hashed_password"), secrets.get("salt")
         ):
             raise AuthFailed(error="Incorrect login or password")
         db.session.add(user_session := UserSession(user_id=query.user.id, token=random_string()))
@@ -120,22 +125,22 @@ async def _get_user_by_token_and_id(id: int, token: str) -> User | None:
         return user
 
     @staticmethod
-    async def register(user_inp: EmailRegister, request: Request) -> ResponseModel | JSONResponse:
+    async def register(user_inp: EmailRegister) -> ResponseModel | JSONResponse:
         confirmation_token: str = random_string()
         auth_method: AuthMethod = (
             db.session.query(AuthMethod)
-            .filter(
+                .filter(
                 AuthMethod.param == "email",
                 func.lower(AuthMethod.value) == user_inp.email.lower(),
                 AuthMethod.auth_method == Email.get_name(),
             )
-            .one_or_none()
+                .one_or_none()
         )
         if auth_method:
             await Email._change_confirmation_link(auth_method.user, confirmation_token)
             send_confirmation_email(
                 to_addr=user_inp.email,
-                link=f"{request.client.host}/email/approve?token={confirmation_token}",
+                link=f"{settings.HOST}/email/approve?token={confirmation_token}",
             )
             return ResponseModel(status="Success", message="Email confirmation link sent")
         if user_inp.user_id and user_inp.token:
@@ -147,7 +152,7 @@ async def register(user_inp: EmailRegister, request: Request) -> ResponseModel |
         await Email._add_to_db(user_inp, confirmation_token, user)
         send_confirmation_email(
             to_addr=user_inp.email,
-            link=f"{request.client.host}/email/approve?token={confirmation_token}",
+            link=f"{settings.HOST}/email/approve?token={confirmation_token}",
         )
         return JSONResponse(
             status_code=201, content=ResponseModel(status="Success", message="Email confirmation link sent").json()
@@ -167,23 +172,23 @@ def validate_password(password: str, hashed_password: str, salt: str):
     async def approve_email(token: str) -> object:
         auth_method = (
             db.session.query(AuthMethod)
-            .filter(
+                .filter(
                 AuthMethod.value == token,
                 AuthMethod.param == "confirmation_token",
                 AuthMethod.auth_method == Email.get_name(),
             )
-            .one_or_none()
+                .one_or_none()
         )
         if not auth_method:
             return JSONResponse(status_code=403, content=ResponseModel(status="Error", message="Incorrect link").json())
         confirmed = (
             db.session.query(AuthMethod)
-            .filter(
+                .filter(
                 AuthMethod.auth_method == Email.get_name(),
                 AuthMethod.param == "confirmed",
                 AuthMethod.user_id == auth_method.user_id,
             )
-            .one()
+                .one()
         )
         confirmed.value = True
         db.session.flush()

auth_backend/base.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel
+from pydantic import BaseModel, constr
 
 
 class Base(BaseModel):
@@ -15,3 +15,7 @@ class Config:
 class ResponseModel(Base):
     status: str
     message: str
+
+
+class Token(Base):
+    token: constr(min_length=1)

auth_backend/routes/base.py

@@ -4,7 +4,7 @@
 
 from auth_backend.auth_plugins.auth_method import AUTH_METHODS
 from auth_backend.settings import get_settings
-from .logout import logout_router
+from .user_session import logout_router
 
 settings = get_settings()
 
@@ -15,6 +15,7 @@
     DBSessionMiddleware,
     db_url=settings.DB_DSN,
     session_args={"autocommit": True},
+    engine_args={"pool_pre_ping": True}
 )
 
 app.add_middleware(

auth_backend/routes/logout.py auth_backend/routes/user_session.pyauth_backend/routes/logout.py renamed to auth_backend/routes/user_session.py

@@ -4,10 +4,9 @@
 from fastapi_sqlalchemy import db
 from starlette.responses import JSONResponse
 
-from auth_backend.base import ResponseModel
-from auth_backend.exceptions import SessionExpired
-
+from auth_backend.base import ResponseModel, Token
 from auth_backend.exceptions import AuthFailed
+from auth_backend.exceptions import SessionExpired, ObjectNotFound
 from auth_backend.models.db import UserSession
 
 logout_router = APIRouter(prefix="", tags=["Logout"])
@@ -23,3 +22,13 @@ async def logout(token: str) -> JSONResponse:
     session.expires = datetime.utcnow()
     db.session.flush()
     return JSONResponse(status_code=200, content=ResponseModel(status="Success", message="Logout successful").dict())
+
+
+@logout_router.post("/{user_id}/token", response_model=ResponseModel)
+async def check_token(user_id: int, token: Token) -> JSONResponse:
+    session = db.session.query(UserSession).filter(UserSession.user_id == user_id, UserSession.token == token.token).one_or_none()
+    if not session:
+        return JSONResponse(status_code=404, content=ResponseModel(status="Error", message="Session not found").json())
+    if session.expired:
+        raise SessionExpired(token.token)
+    return JSONResponse(status_code=200, content=ResponseModel(status="Success", message="Session found and exists").json())

auth_backend/settings.py

@@ -7,6 +7,7 @@ class Settings(BaseSettings):
     DB_DSN: PostgresDsn
 
     EMAIL: str | None
+    HOST: str = "localhost"
     EMAIL_PASS: str = None
     SMTP_HOST: str = 'smtp.gmail.com'
     SMTP_PORT: int = 587

tests/test_routes/test_login.py

@@ -77,3 +77,30 @@ def test_incorrect_data(client: TestClient, dbsession: Session):
         dbsession.delete(row)
     dbsession.delete(dbsession.query(User).filter(User.id == id).one())
     dbsession.flush()
+
+
+
+def test_check_token(client: TestClient, user, dbsession: Session):
+    user_id, body, login = user["user_id"], user["body"], user["login_json"]
+
+    response = client.post(f"/{user_id}/token", json={"token": login["token"] + "2"})
+    assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    response = client.post(f"/{user_id}/token", json={"token": login["token"]})
+    assert response.status_code == status.HTTP_200_OK
+
+    response = client.post(f"/logout?token={login['token']}")
+    assert response.status_code == status.HTTP_200_OK
+
+    response = client.post(f"/{user_id}/token", json={"token": login["token"]})
+    assert response.status_code == status.HTTP_403_FORBIDDEN
+
+
+def test_invalid_check_tokens(client: TestClient, user):
+    user_id, body, login = user["user_id"], user["body"], user["login_json"]
+    response = client.post(f"/{user_id}/token", json={"token": ""})
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+
+    response = client.post(f"/{user_id}/token")
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
+

tests/test_routes/test_registration.py

@@ -24,13 +24,17 @@ def test_invalid_email(client: TestClient):
         "password": "&%@#$@322îïíīįì3@##EFWed}efvef{}{}{}[èéêëēėę'"
     }
     body4 = {
-        "email": f"EmailFor+#&*|_ Sur{datetime.datetime.utcnow()}e@mail.gtg",
+        "email": f"EmailFor+ _Sur{datetime.datetime.utcnow()}e@mail.gtg",
         "password": "string2222"
     }
     body5 = {
         "email": f"Email For Sure {datetime.datetime.utcnow()} @ mail. gtg",
         "password": "string"
     }
+    body6 = {
+        "email": f"roman@dyakov.space\nContent-Type: text/html; charset=utf-8;\n\nАхаха,лох<!---",
+        "password": "string"
+    }
     response = client.post(url, json=body1)
     assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
     response = client.post(url, json=body2)
@@ -41,7 +45,8 @@ def test_invalid_email(client: TestClient):
     assert response.status_code == status.HTTP_201_CREATED
     response = client.post(url, json=body5)
     assert response.status_code == status.HTTP_201_CREATED
-
+    response = client.post(url, json=body6)
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
 
 
 def test_main_scenario(client: TestClient, dbsession: Session):