feat: added JWT-based authentication; introduced config file

* authorization [WIP]
* refactored several sections
* used dict(exclude_unset=True)
* global values are now sourced from a config file (statuspagerc)

app/auth.py

@@ -0,0 +1,42 @@
+import bcrypt
+import jwt
+from fastapi import HTTPException
+from datetime import datetime, timedelta
+from app import crud
+from app.database import SessionLocal
+from app.settings import globals as settings
+
+
+
+KEY = settings.JWT_SECRET
+ALGORITHM = settings.JWT_ALGORITHM
+EXPIRY = settings.JWT_EXPIRY
+
+
+def login(username: str, password: str):
+	db = SessionLocal()
+	usr = crud.login_info(db=db, name=username,password=password)
+	if usr is None:
+		return {"status": "error", "mesage": "username/password incorrect"}
+	description = 'access_token'
+	token = jwt.encode({
+		'sub': username,
+		'iat': datetime.utcnow(),
+		'exp': datetime.utcnow() + timedelta(seconds=EXPIRY),
+		'des': description
+	},
+	KEY, ALGORITHM)
+	return {"status": "success", "token": token.decode('utf-8')}
+
+def validate(token):
+	try:
+		data = jwt.decode(token, KEY)
+	except Exception as e:
+		if "expired" in str(e):
+			raise HTTPException(status_code=401, detail={"status": "error", "message": "Token expired"})
+		elif "Not enough segments" in str(e):
+			raise HTTPException(status_code=401, detail={"status": "error", "message": "Invalid token"})
+		else:
+			raise HTTPException(status_code=400, detail={"status": "error", "message": "Exception: " + str(e)})
+	return data
+

app/crud.py

@@ -1,6 +1,6 @@
 import bcrypt, time
 from sqlalchemy.orm import Session
-from . import models, schemas
+from app import models, schemas
 
 
 User=models.User
@@ -34,7 +34,7 @@ def create_user(db: Session, user: schemas.UserCreate):
 
 
 def update_user(db: Session, user_id: int, user: schemas.UserUpdate):
-	updated_user = User(**user.dict())
+	updated_user = User(**user.dict(exclude_unset=True))
 	if (updated_user.password is not None):
 		updated_user.password = bcrypt.hashpw(user.password.encode(), bcrypt.gensalt())
 		db.query(User).filter(User.id==user_id).update({User.password: updated_user.password})
@@ -74,7 +74,7 @@ def del_service(db: Session, service_id: int):
 
 
 def update_service(db: Session, service_id: int, s: schemas.ServiceUpdate):
-	updated_s = Service(**s.dict())
+	updated_s = Service(**s.dict(exclude_unset=True))
 	if (updated_s.name is not None):
 		db.query(Service).filter(Service.id==service_id).update({Service.name: updated_s.name})
 	if (updated_s.owner_id is not None):
@@ -93,3 +93,19 @@ def update_service(db: Session, service_id: int, s: schemas.ServiceUpdate):
 		db.query(Service).filter(Service.id==service_id).update({Service.updated_unix: time.time()})
 	return db.commit()
 
+
+
+
+def login_info(db: Session, name: str, password: str):
+	try:
+		usr = db.query(User).filter(User.name == name).first()
+		if bcrypt.hashpw(password.encode(), usr.password) == usr.password:
+			print("[*] match")
+		else:
+			print("[x] no match")
+			return None
+	except Exception as e:
+		print('exception', e)
+		return
+	return usr
+

app/main.py

@@ -3,16 +3,18 @@
 
 from typing import List, Optional
 
-from fastapi import FastAPI, Depends, HTTPException
+from fastapi import FastAPI, Depends, Security, HTTPException
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from sqlalchemy.orm import Session
 
-from app import crud, models, schemas
+from app import crud, models, schemas, auth
 from app.database import SessionLocal, engine
 
 
 models.Base.metadata.create_all(bind=engine)
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/v1/auth/get_token')
 
-app = FastAPI()
+app = FastAPI(title="Statuspage API",description="This documentation describes the Statuspage API.",version="0.0.1")
 
 def get_db():
 	db = SessionLocal()
@@ -21,7 +23,10 @@ def get_db():
 	finally:
 		db.close()
 
-
+def is_auth(token: str = Security(oauth2_scheme)):
+	if not token:
+		raise HTTPException(status_code=401, detail={"status": "error", "message": "Token not provided "})
+	return auth.validate(token)
 
 
 def get_usr(user_id: int, db: Session):
@@ -49,7 +54,7 @@ async def read_services(skip: int = 0, limit: int = 100, db: Session = Depends(g
 	return crud.get_services(db, skip=skip, limit=limit)
 
 @app.post("/api/v1/service", response_model=schemas.Service)
-async def create_service(service: schemas.ServiceCreate, db: Session = Depends(get_db)):
+async def create_service(service: schemas.ServiceCreate, db: Session = Depends(get_db), token: str = Security(is_auth)):
 	"""Service types: icmp = 0, http = 1"""
 	if service.service_type not in [0, 1]:
 		raise HTTPException(status_code=422, detail="Invalid service type provided")
@@ -66,7 +71,7 @@ async def read_service(service_id: int, db: Session = Depends(get_db)):
 	return db_service
 
 @app.patch("/api/v1/service/{service_id}")
-async def update_service(service_id: int, service: schemas.ServiceUpdate, db: Session = Depends(get_db)):
+async def update_service(service_id: int, service: schemas.ServiceUpdate, db: Session = Depends(get_db), token: str = Security(is_auth)):
 	if ((service.name is None) and (service.owner_id is None) and (service.is_private is None) and (service.description is None) and (service.service_type is None) and (service.url is None) and (service.is_active is None)):
 		raise HTTPException(status_code=400, detail="No data provided")
 	db_service = get_srv(service_id, db)
@@ -76,7 +81,7 @@ async def update_service(service_id: int, service: schemas.ServiceUpdate, db: Se
 	return {"update":"success"}
 
 @app.delete("/api/v1/service/{service_id}")
-async def delete_service(service_id: int, db: Session = Depends(get_db)):
+async def delete_service(service_id: int, db: Session = Depends(get_db), token: str = Security(is_auth)):
 	db_service = get_srv(service_id, db)
 	if db_service is None:
 		raise HTTPException(status_code=422, detail="Unprocessable entity")
@@ -86,7 +91,7 @@ async def delete_service(service_id: int, db: Session = Depends(get_db)):
 
 
 @app.post("/api/v1/users", response_model=schemas.User)
-async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
+async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db), token: str = Security(is_auth)):
 	db_user = crud.get_user_by_username(db, name=user.name)
 	if db_user:
 		raise HTTPException(status_code=400, detail="Username already registered")
@@ -105,7 +110,7 @@ async def read_user(user_id: int, db: Session = Depends(get_db)):
 	return db_user
 
 @app.patch("/api/v1/users/{user_id}")
-async def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):
+async def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db), token: str = Security(is_auth)):
 	if ((user.name is None) and (user.password is None) and (user.full_name is None) and (user.is_active is None)):
 		raise HTTPException(status_code=400, detail="No data provided")
 	db_user = get_usr(user_id, db)
@@ -115,10 +120,25 @@ async def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depe
 	return {"update":"success"}
 
 @app.delete("/api/v1/users/{user_id}")
-async def delete_user(user_id: int, db: Session = Depends(get_db)):
+async def delete_user(user_id: int, db: Session = Depends(get_db), token: str = Security(is_auth)):
 	db_user = get_usr(user_id, db)
 	if db_user is None:
 		raise HTTPException(status_code=422, detail="Unprocessable entity")
 	crud.del_user(db=db, user_id=user_id)
 	return {"delete":"success"}
 
+
+
+@app.post('/api/v1/auth/get_token',response_model=schemas.Token,response_description="Returns user access token",summary="Authenticate API user",description="Authenticate an API user and return a token for subsequent requests")
+async def get_token(form_data: OAuth2PasswordRequestForm = Depends()):
+	a = auth.login(form_data.username, form_data.password)
+	if a and a["status"] == "error":
+		raise HTTPException(status_code=400, detail={"status": "error", "mesage": "username/password incorrect"})
+	return {"access_token": a["token"], "token_type": "bearer"}
+
+@app.post('/api/v1/auth/validate')
+async def validate_token(token: str):
+	a = auth.validate(token=token)
+	if a:
+		return a
+

app/schemas.py

@@ -70,3 +70,7 @@ class APISimpleErrorResponse(BaseModel):
 	success: bool = False
 	errors: Optional[List[str]]
 
+class Token(BaseModel):
+	access_token: str
+	token_type: str
+

app/settings/globals.py

@@ -0,0 +1,19 @@
+from pathlib import Path
+from typing import Optional
+
+from starlette.config import Config
+
+
+p: Path = Path(__file__).parents[2] / "statuspagerc"
+config: Config = Config(p if p.exists() else None)
+
+DATABASE: str = config("DATABASE", cast=str)
+
+ALEMBIC_CONFIG: str = (
+	DATABASE
+)
+
+JWT_SECRET: str = config("JWT_SECRET", cast=str)
+JWT_ALGORITHM: str = config("JWT_ALGORITHM", cast=str)
+JWT_EXPIRY: int = config("JWT_EXPIRY", cast=int)
+

statuspagerc

@@ -0,0 +1,12 @@
+# statuspage configuration file
+
+# database file
+DATABASE=sqlite:///sql_app.db
+
+# [JWT related parameters]
+# it's imperative you CHANGE JWT_SECRET to a unique value
+# you can generate a random one with 'openssl rand -hex 64'
+JWT_SECRET = 'deadbeefdeafbeefd34db33fdeadbeefdeafbeefd34db33fdeadbeefdeafbeefd34db33f'
+JWT_ALGORITHM = "HS256"
+# token expiration time in seconds
+JWT_EXPIRY = 7200