Feature/login

.gitignore

@@ -142,6 +142,9 @@ dmypy.json
 # Pyre type checker
 .pyre/
 
+
+# register stuff
+run.txt
 # VScode
 .vscode/
 app/.vscode/

README.md

@@ -44,6 +44,22 @@ uvicorn app.main:app --reload
 python -m pytest --cov-report term-missing --cov=app tests
 ```
 
+### Generating  personal JWT Secret Key
+'''shell
+python
+import secrets
+secrets.token_hex(32)
+# copy generated string.
+# in config.py file, replace JWT_KEY value with generated string
+''' 
+
+## Using security dependencies:
+'''
+from app.internal.security.dependencies:
+use current_user_required and current_user functions as a route dependencies.
+an example for how to use can be found at tests.security_testing_routes file.
+'''
+
 ## Contributing
 View [contributing guidelines](https://github.com/PythonFreeCourse/calendar/blob/master/CONTRIBUTING.md).
 

app/config.py.example

@@ -48,6 +48,13 @@ email_conf = ConnectionConfig(
     USE_CREDENTIALS=True,
 )
 
+# register
+DATABASE_CONNECTION_STRING = "sqlite:///calendar.db"
+
+# security
+JWT_KEY = "c06857128217c661de43d746ecdbe9f1dbc3b3685957fab64512456f639b1881"
+JWT_ALGORITHM = "HS256"
+JWT_MIN_EXP = 60 * 24 * 7
 templates = Jinja2Templates(directory=os.path.join("app", "templates"))
 
 # application name

app/database/schemas.py

@@ -0,0 +1,90 @@
+from typing import Optional, Union
+from pydantic import BaseModel, validator, EmailStr, EmailError
+
+
+EMPTY_FIELD_STRING = 'field is required'
+MIN_FIELD_LENGTH = 3
+MAX_FIELD_LENGTH = 20
+
+
+def fields_not_empty(field: Optional[str]) -> Union[ValueError, str]:
+    """Global function to validate fields are not empty."""
+    if not field:
+        raise ValueError(EMPTY_FIELD_STRING)
+    return field
+
+
+class UserBase(BaseModel):
+    """
+    Validating fields types
+    Returns a User object without sensitive information
+    """
+    username: str
+    email: str
+    full_name: str
+    description: Optional[str] = None
+
+    class Config:
+        orm_mode = True
+
+
+class UserCreate(UserBase):
+    """Validating fields types"""
+    password: str
+    confirm_password: str
+
+    """
+    Calling to field_not_empty validaion function,
+    for each required field.
+    """
+    _fields_not_empty_username = validator(
+        'username', allow_reuse=True)(fields_not_empty)
+    _fields_not_empty_full_name = validator(
+        'full_name', allow_reuse=True)(fields_not_empty)
+    _fields_not_empty_password = validator(
+        'password', allow_reuse=True)(fields_not_empty)
+    _fields_not_empty_confirm_password = validator(
+        'confirm_password', allow_reuse=True)(fields_not_empty)
+    _fields_not_empty_email = validator(
+        'email', allow_reuse=True)(fields_not_empty)
+
+    @validator('confirm_password')
+    def passwords_match(
+            cls, confirm_password: str,
+            values: UserBase) -> Union[ValueError, str]:
+        """Validating passwords fields identical."""
+        if 'password' in values and confirm_password != values['password']:
+            raise ValueError("doesn't match to password")
+        return confirm_password
+
+    @validator('username')
+    def username_length(cls, username: str) -> Union[ValueError, str]:
+        """Validating username length is legal"""
+        if not (MIN_FIELD_LENGTH < len(username) < MAX_FIELD_LENGTH):
+            raise ValueError("must contain between 3 to 20 charactars")
+        return username
+
+    @validator('password')
+    def password_length(cls, password: str) -> Union[ValueError, str]:
+        """Validating username length is legal"""
+        if not (MIN_FIELD_LENGTH < len(password) < MAX_FIELD_LENGTH):
+            raise ValueError("must contain between 3 to 20 charactars")
+        return password
+
+    @validator('email')
+    def confirm_mail(cls, email: str) -> Union[ValueError, str]:
+        """Validating email is valid mail address."""
+        try:
+            EmailStr.validate(email)
+            return email
+        except EmailError:
+            raise ValueError("address is not valid")
+
+
+class User(UserBase):
+    """
+    Validating fields types
+    Returns a User object without sensitive information
+    """
+    id: int
+    is_active: bool

app/internal/security/dependancies.py

@@ -0,0 +1,33 @@
+from typing import Union
+
+from app.database.database import get_db
+from app.internal.security.ouath2 import (
+    Depends, Session, check_jwt_token, get_cookie)
+from app.internal.security.schema import CurrentUser
+from starlette.requests import Request
+
+
+async def current_user_required(
+        request: Request,
+        db: Session = Depends(get_db),
+        jwt: str = Depends(get_cookie)) -> CurrentUser:
+    """A dependency function protecting routes for only logged in user"""
+    user = await check_jwt_token(db, jwt, path=request.url.path)
+    if user:
+        return user
+
+
+async def current_user(
+    request: Request,
+        db: Session = Depends(get_db),) -> Union[CurrentUser, bool]:
+    """
+    A dependency function.
+    Returns logged in User object if exists.
+    None if not.
+    """
+    if 'Authorization' in request.cookies:
+        jwt = request.cookies['Authorization']
+    else:
+        return None
+    user = await check_jwt_token(db, jwt)
+    return user

app/internal/security/ouath2.py

@@ -0,0 +1,127 @@
+from datetime import datetime, timedelta
+from typing import Union
+
+from app.config import JWT_ALGORITHM, JWT_KEY, JWT_MIN_EXP
+from passlib.context import CryptContext
+from app.database.models import User
+from fastapi import Depends, HTTPException
+from fastapi.security import OAuth2PasswordBearer
+import jwt
+from jwt.exceptions import InvalidSignatureError
+from sqlalchemy.orm import Session
+from starlette.requests import Request
+from starlette.responses import RedirectResponse
+from starlette.status import HTTP_401_UNAUTHORIZED
+from .schema import LoginUser, CurrentUser
+
+
+pwd_context = CryptContext(schemes=["bcrypt"])
+oauth_schema = OAuth2PasswordBearer(tokenUrl="/login")
+
+
+async def get_db_user_by_username(db: Session, username: str) -> User:
+    """Checking database for user by username field"""
+    user = db.query(User).filter_by(username=username).first()
+    return user
+
+
+def get_hashed_password(password) -> str:
+    """Hashing user password"""
+    return pwd_context.hash(password)
+
+
+def verify_password(plain_password, hashed_password) -> bool:
+    """Verifying password and hashed password are equal"""
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+async def authenticate_user(
+        db: Session, user: LoginUser) -> Union[LoginUser, bool]:
+    """Verifying user is in database and password is correct"""
+    db_user = await get_db_user_by_username(db=db, username=user.username)
+    if db_user:
+        if verify_password(user.password, db_user.password):
+            return LoginUser(
+                username=user.username, password=db_user.password)
+    return False
+
+
+def create_jwt_token(
+        user: LoginUser, JWT_MIN_EXP: int = JWT_MIN_EXP,
+        JWT_KEY: str = JWT_KEY) -> str:
+    """Creating jwt-token out of user unique data"""
+    expiration = datetime.utcnow() + timedelta(minutes=JWT_MIN_EXP)
+    jwt_payload = {
+        "sub": user.username,
+        "hashed_password": user.password,
+        "exp": expiration}
+    jwt_token = jwt.encode(
+        jwt_payload, JWT_KEY, algorithm=JWT_ALGORITHM)
+    return jwt_token
+
+
+async def check_jwt_token(
+    db: Session,
+    token: str = Depends(oauth_schema),
+        path: bool = None) -> CurrentUser:
+    """
+    Check whether JWT token is correct. Returns User object if yes.
+    Returns None or raises HTTPException,
+    depanding which depandency activated this function.
+    """
+    try:
+        jwt_payload = jwt.decode(
+            token, JWT_KEY, algorithms=JWT_ALGORITHM)
+        jwt_username = jwt_payload.get("sub")
+        jwt_hashed_password = jwt_payload.get("hashed_password")
+        db_user = await get_db_user_by_username(db, username=jwt_username)
+        if db_user and db_user.password == jwt_hashed_password:
+            return CurrentUser(**db_user.__dict__)
+        else:
+            raise HTTPException(
+                status_code=HTTP_401_UNAUTHORIZED,
+                headers=path,
+                detail="Your token is incorrect. Please log in again")
+    except InvalidSignatureError:
+        raise HTTPException(
+                status_code=HTTP_401_UNAUTHORIZED,
+                headers=path,
+                detail="Your token is incorrect. Please log in again")
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(
+            status_code=HTTP_401_UNAUTHORIZED,
+            headers=path,
+            detail="Your token has expired. Please log in again")
+    except jwt.DecodeError:
+        raise HTTPException(
+            status_code=HTTP_401_UNAUTHORIZED,
+            headers=path,
+            detail="Your token is incorrect. Please log in again")
+
+
+async def get_cookie(request: Request) -> str:
+    """
+    Extracts jwt from HTTPONLY cookie, if exists.
+    Raises HTTPException if not.
+    """
+    if 'Authorization' in request.cookies:
+        return request.cookies['Authorization']
+    raise HTTPException(
+        status_code=HTTP_401_UNAUTHORIZED,
+        headers=request.url.path,
+        detail="Please log in to enter this page")
+
+
+async def my_exception_handler(
+        request: Request,
+        exc: HTTP_401_UNAUTHORIZED) -> RedirectResponse:
+    """
+    Whenever HTTP_401_UNAUTHORIZED is raised,
+    redirecting to login route, with original requested url,
+    and details for why original request failed.
+    """
+    paramas = f"?next={exc.headers}&message={exc.detail}"
+    url = f"/login{paramas}"
+    response = RedirectResponse(url=url)
+    response.delete_cookie('Authorization')
+    return response

app/internal/security/schema.py

@@ -0,0 +1,37 @@
+from typing import List, Optional
+
+from app.database.models import UserEvent
+from pydantic import BaseModel
+
+
+class LoginUser(BaseModel):
+    """
+    Validating fields types
+    Returns a User object for signing in.
+    """
+    username: str
+    password: str
+
+    class Config:
+        orm_mode = True
+
+
+class CurrentUser(BaseModel):
+    """
+    Security dependencies will return this object,
+    instead of db object.
+    Returns all User's parameters, except password.
+    """
+    id: int
+    username: str
+    full_name: str
+    email: str
+    language: str = None
+    description: str = None
+    avatar: str
+    telegram_id: str = None
+    events = Optional[List[UserEvent]]
+
+    class Config:
+        orm_mode = True
+        arbitrary_types_allowed = True

app/internal/user.py

@@ -0,0 +1,46 @@
+from app.database import models, schemas
+from app.internal.security.ouath2 import get_hashed_password
+from sqlalchemy.orm import Session
+
+
+def get_by_id(db: Session, user_id: int) -> models.User:
+    """query database for a user by unique id"""
+    return db.query(models.User).filter(models.User.id == user_id).first()
+
+
+def get_by_username(db: Session, username: str) -> models.User:
+    """query database for a user by unique username"""
+    return db.query(models.User).filter(
+        models.User.username == username).first()
+
+
+def get_by_mail(db: Session, email: str) -> models.User:
+    """query database for a user by unique email"""
+    return db.query(models.User).filter(models.User.email == email).first()
+
+
+def create(db: Session, user: schemas.UserCreate) -> models.User:
+    """
+    creating a new User object in the database, with hashed password
+    """
+    unhashed_password = user.password.encode('utf-8')
+    hashed_password = get_hashed_password(unhashed_password)
+    user_details = {
+        'username': user.username,
+        'full_name': user.full_name,
+        'email': user.email,
+        'password': hashed_password,
+        'description': user.description
+    }
+    db_user = models.User(**user_details)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+
+def delete_by_mail(db: Session, email: str) -> None:
+    """deletes a user from database by unique email"""
+    db_user = get_by_mail(db=db, email=email)
+    db.delete(db_user)
+    db.commit()