diff --git a/app/__init__.py b/app/__init__.py index 1217788..e4c3274 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -4,4 +4,4 @@ from .controllers import cardManager, userManager app = FastAPI() app.include_router(userManager.user_router) -app.include_router(cardManager.card_router) \ No newline at end of file +app.include_router(cardManager.card_router) diff --git a/app/controllers/aaManager.py b/app/controllers/aaManager.py index ad31e64..2a5f49c 100644 --- a/app/controllers/aaManager.py +++ b/app/controllers/aaManager.py @@ -12,40 +12,65 @@ from app.services.database import add_and_refresh, get_session aa_router = APIRouter(prefix="/api/v1/aa", tags=["AccessAuth"]) + @aa_router.post("/", response_model=AccessAuthorizationResponse) -def add_accessauth(*, db: Session = Depends(get_session), aa: AccessAuthorizationCreate, admin: bool = Depends(auth_is_admin)): +def add_accessauth( + *, + db: Session = Depends(get_session), + aa: AccessAuthorizationCreate, + admin: bool = Depends(auth_is_admin), +): logger.info(f"Creating accessauth with data: {aa}") if aa.timetables != []: timetables = [Timetable.model_validate(t) for t in aa.timetables] - else: timetables = [] + else: + timetables = [] if aa.oneshot is not None: oneshot = OneShotAccess.model_validate(aa.oneshot) - else: oneshot = None + else: + oneshot = None db_aa = AccessAuthorizationDB( name=aa.name, type=aa.type, is_active=aa.is_active, timetables=timetables, - oneshot=oneshot + oneshot=oneshot, ) return add_and_refresh(db, db_aa) + @aa_router.get("/", response_model=list[AccessAuthorizationResponse]) -def get_all_accessauths(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def get_all_accessauths( + db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): return db.exec( - select(AccessAuthorizationDB) - .options(selectinload(AccessAuthorizationDB.timetables)) - ).all() + select(AccessAuthorizationDB).options( + selectinload(AccessAuthorizationDB.timetables) + ) + ).all() + @aa_router.get("/{aa_id}", response_model=AccessAuthorizationResponse) -def get_one_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)): +def get_one_accessauth( + *, + db: Session = Depends(get_session), + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AA not found") return db_aa + @aa_router.put("/assign/{group_id}/{aa_id}", response_model=GroupResponse) -def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int, admin: bool = Depends(auth_is_admin)): +def assign_accessauth( + *, + db: Session = Depends(get_session), + group_id: int, + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_group = db.get(GroupDB, group_id) if db_group is None: raise HTTPException(status_code=404, detail="Group not found") @@ -53,12 +78,21 @@ def assign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_i if db_aa is None: raise HTTPException(status_code=404, detail="AA not found") if db_aa in db_group.accessauths: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail="AA already assigned to group" + ) db_group.accessauths.append(db_aa) return add_and_refresh(db, db_group) + @aa_router.put("/unassign/{group_id}/{aa_id}", response_model=GroupResponse) -def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa_id: int, admin: bool = Depends(auth_is_admin)): +def unassign_accessauth( + *, + db: Session = Depends(get_session), + group_id: int, + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_group = db.get(GroupDB, group_id) if db_group is None: raise HTTPException(status_code=404, detail="Group not found") @@ -70,8 +104,15 @@ def unassign_accessauth(*, db: Session = Depends(get_session), group_id: int, aa db_group.accessauths.remove(db_aa) return add_and_refresh(db, db_group) + @aa_router.patch("/{aa_id}", response_model=AccessAuthorizationResponse) -def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: AccessAuthorizationUpdate, admin: bool = Depends(auth_is_admin)): +def change_accessauth( + *, + db: Session = Depends(get_session), + aa_id: int, + aa: AccessAuthorizationUpdate, + admin: bool = Depends(auth_is_admin), +): db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AccessAuthorization not found") @@ -88,11 +129,17 @@ def change_accessauth(*, db: Session = Depends(get_session), aa_id: int, aa: Acc db_aa.sqlmodel_update(aa_data) return add_and_refresh(db, db_aa) + @aa_router.delete("/{aa_id}") -def delete_accessauth(*, db: Session = Depends(get_session), aa_id: int, admin: bool = Depends(auth_is_admin)): +def delete_accessauth( + *, + db: Session = Depends(get_session), + aa_id: int, + admin: bool = Depends(auth_is_admin), +): db_aa = db.get(AccessAuthorizationDB, aa_id) if db_aa is None: raise HTTPException(status_code=404, detail="AccessAuthorization not found") db.delete(db_aa) db.commit() - return {"message": "AccessAuthorization deleted successfully"} \ No newline at end of file + return {"message": "AccessAuthorization deleted successfully"} diff --git a/app/controllers/cardManager.py b/app/controllers/cardManager.py index 68cba9d..172c098 100644 --- a/app/controllers/cardManager.py +++ b/app/controllers/cardManager.py @@ -13,58 +13,100 @@ from app.services.scanner import DeleteCard, WriteNewCard card_router = APIRouter(prefix="/api/v1/cards", tags=["Card"]) + def register_card(cardInput: CardCreate): key, uid = WriteNewCard() if key == None: logger.info("No card registered. Check logs!") - raise HTTPException(status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!") + raise HTTPException( + status.HTTP_417_EXPECTATION_FAILED, detail="No card registered. Check logs!" + ) card = Card( group_id=cardInput.group_id, key=key, name=cardInput.name, card_serial=uid, - enabled=cardInput.enabled - ) + enabled=cardInput.enabled, + ) return card + @card_router.post("/", response_model=Card) -def add_card(cardInput: CardCreate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): - try: assert db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() == None +def add_card( + cardInput: CardCreate, + db: Session = Depends(get_session), + admin: bool = Depends(auth_is_admin), +): + try: + assert ( + db.exec(select(Card).where(Card.name == cardInput.name)).one_or_none() + == None + ) except AssertionError: raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!") - try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None + try: + assert ( + db.exec( + select(GroupDB).where(GroupDB.id == cardInput.group_id) + ).one_or_none() + is not None + ) except AssertionError: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!") card = register_card(cardInput) return add_and_refresh(db, card) + @card_router.delete("/") -def del_card(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def del_card( + *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): key = DeleteCard() logger.info(key) try: card = db.exec(select(Card).where(Card.key == key)).one() except NoResultFound: logger.info(f"The key:'{key}' was not found in db!") - raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Key on card not found in DB. Please tell an admin about this. KEY={key}") + raise HTTPException( + status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Key on card not found in DB. Please tell an admin about this. KEY={key}", + ) db.delete(card) db.commit() return {"message": "Card deleted successfully"} + @card_router.get("/{group_id}", response_model=list[Card]) -def get_cards(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)): +def get_cards( + *, + db: Session = Depends(get_session), + group_id: int, + admin: bool = Depends(auth_is_admin), +): cards = db.exec(select(Card).where(Card.group_id == group_id)).all() return cards + @card_router.patch("/{card_id}", response_model=Card) -def update_card(card_id: int, cardInput: CardUpdate, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def update_card( + card_id: int, + cardInput: CardUpdate, + db: Session = Depends(get_session), + admin: bool = Depends(auth_is_admin), +): db_card = db.get(Card, card_id) if db_card is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Card not found!") card_data = cardInput.model_dump(exclude_unset=True, exclude_none=True) if "group_id" in card_data: - try: assert db.exec(select(GroupDB).where(GroupDB.id == cardInput.group_id)).one_or_none() is not None + try: + assert ( + db.exec( + select(GroupDB).where(GroupDB.id == cardInput.group_id) + ).one_or_none() + is not None + ) except AssertionError: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="GroupID not found!") db_card.sqlmodel_update(card_data) - return add_and_refresh(db, db_card) \ No newline at end of file + return add_and_refresh(db, db_card) diff --git a/app/controllers/debugManager.py b/app/controllers/debugManager.py index 0f6addf..fd5b49a 100644 --- a/app/controllers/debugManager.py +++ b/app/controllers/debugManager.py @@ -15,20 +15,30 @@ debug_router = APIRouter( dependencies=[Depends(auth_is_admin)], ) + @debug_router.put("/addcard/") -def add_card_manually(groupid: int, card_key: str, name: str, enabled: bool, db: Session=Depends(get_session)): +def add_card_manually( + groupid: int, + card_key: str, + name: str, + enabled: bool, + db: Session = Depends(get_session), +): """Add cards manually (you also have to delete them manually)""" - logger.critical(f"Manual db change: adding a card with key: {card_key} to group: {groupid}") + logger.critical( + f"Manual db change: adding a card with key: {card_key} to group: {groupid}" + ) card = Card( group_id=groupid, key=card_key, name=name, enabled=enabled, - card_serial="00:00:00:00:00:00:00" - ) + card_serial="00:00:00:00:00:00:00", + ) add_and_refresh(db, card) return card + @debug_router.get("/rmcard/{card_id}") def remove_card_manually(card_id: str, db: Session = Depends(get_session)): try: @@ -39,7 +49,7 @@ def remove_card_manually(card_id: str, db: Session = Depends(get_session)): db.delete(card) db.commit() return {"message": "Card deleted successfully"} - + @debug_router.put("/getcards") def list_all_cards(db: Session = Depends(get_session)): @@ -49,4 +59,4 @@ def list_all_cards(db: Session = Depends(get_session)): out = [] for i in cards: out.append({i.key: i.group.name}) - return out \ No newline at end of file + return out diff --git a/app/controllers/doorManager.py b/app/controllers/doorManager.py index bfe22d0..88cc8c2 100644 --- a/app/controllers/doorManager.py +++ b/app/controllers/doorManager.py @@ -5,16 +5,19 @@ import app.services.door as doorService from app.services.auth import auth_is_admin from app.services.database import get_session -door_router = APIRouter(prefix="/api/v1/door",tags=["Door"]) +door_router = APIRouter(prefix="/api/v1/door", tags=["Door"]) + @door_router.put("/open") def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): doorService.opendoor() + @door_router.put("/close") def open_door(db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): doorService.closedoor() + @door_router.post("/test") def test_access(input: str, db: Session = Depends(get_session)): - return doorService.checkAccess(input, db=db) \ No newline at end of file + return doorService.checkAccess(input, db=db) diff --git a/app/controllers/groupManager.py b/app/controllers/groupManager.py index 88a6786..0b76b51 100644 --- a/app/controllers/groupManager.py +++ b/app/controllers/groupManager.py @@ -1,4 +1,3 @@ - from fastapi import APIRouter, Depends, HTTPException, status from sqlmodel import Session, select @@ -8,24 +7,41 @@ from ..services.database import add_and_refresh, get_session group_router = APIRouter(prefix="/api/v1/groups", tags=["Group"]) + @group_router.get("/", response_model=list[GroupResponse]) -def get_groups(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def get_groups( + *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): groups = db.exec(select(GroupDB)).all() return groups + @group_router.post("/", response_model=GroupResponse) -def create_group(*, db: Session = Depends(get_session), group: GroupCreate, admin: bool = Depends(auth_is_admin)): +def create_group( + *, + db: Session = Depends(get_session), + group: GroupCreate, + admin: bool = Depends(auth_is_admin), +): db_group = GroupDB.model_validate(group) group = db.exec(select(GroupDB).where(GroupDB.name == db_group.name)).first() if group is not None: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Group already exists!") + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail="Group already exists!" + ) return add_and_refresh(db, db_group) + @group_router.delete("/{group_id}") -def delete_group(*, db: Session = Depends(get_session), group_id: int, admin: bool = Depends(auth_is_admin)): +def delete_group( + *, + db: Session = Depends(get_session), + group_id: int, + admin: bool = Depends(auth_is_admin), +): db_group = db.get(GroupDB, group_id) if db_group is None: raise HTTPException(status_code=404, detail="Group not found") db.delete(db_group) db.commit() - return {"message": "Group deleted successfully"} \ No newline at end of file + return {"message": "Group deleted successfully"} diff --git a/app/controllers/userManager.py b/app/controllers/userManager.py index f19a753..9a63813 100644 --- a/app/controllers/userManager.py +++ b/app/controllers/userManager.py @@ -12,33 +12,62 @@ from ..services.database import add_and_refresh, get_session user_router = APIRouter(tags=["Users"], prefix="/api/v1/users") + @user_router.post("/", response_model=UserResponse) -def create_user(*, db: Session = Depends(get_session), user: UserCreate, admin: bool = Depends(auth_is_admin)): +def create_user( + *, + db: Session = Depends(get_session), + user: UserCreate, + admin: bool = Depends(auth_is_admin), +): hashed_password = {"passwordhash": get_password_hash(user.password)} - try: assert db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() == None + try: + assert ( + db.exec(select(UserDB).where(UserDB.name == user.name)).one_or_none() + == None + ) except AssertionError: raise HTTPException(status.HTTP_409_CONFLICT, detail="Name already used!") db_user = UserDB.model_validate(user, update=hashed_password) return add_and_refresh(db, db_user) + @user_router.get("/", response_model=list[UserResponse]) -def read_users(*, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin)): +def read_users( + *, db: Session = Depends(get_session), admin: bool = Depends(auth_is_admin) +): users = db.exec(select(UserDB)).all() return users + @user_router.get("/current", response_model=UserResponse) -def get_current_user(db: Session = Depends(get_session), user: UserDB = Depends(auth_user)): +def get_current_user( + db: Session = Depends(get_session), user: UserDB = Depends(auth_user) +): return user + @user_router.get("/{user_id}", response_model=UserResponse) -def read_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)): +def read_user( + *, + db: Session = Depends(get_session), + user_id: int, + admin: bool = Depends(auth_is_admin), +): db_user = db.get(UserDB, user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") return db_user + @user_router.patch("/{user_id}", response_model=UserResponse) -def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserUpdate, admin: bool = Depends(auth_is_admin)): +def update_user( + *, + db: Session = Depends(get_session), + user_id: int, + user: UserUpdate, + admin: bool = Depends(auth_is_admin), +): db_user = db.get(UserDB, user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") @@ -50,12 +79,17 @@ def update_user(*, db: Session = Depends(get_session), user_id: int, user: UserU db_user.sqlmodel_update(user_data, update=hashed_password) return add_and_refresh(db, db_user) + @user_router.delete("/{user_id}") -def delete_user(*, db: Session = Depends(get_session), user_id: int, admin: bool = Depends(auth_is_admin)): +def delete_user( + *, + db: Session = Depends(get_session), + user_id: int, + admin: bool = Depends(auth_is_admin), +): db_user = db.get(UserDB, user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") db.delete(db_user) db.commit() return {"message": "User deleted successfully"} - diff --git a/app/main.py b/app/main.py index 3e6dde4..ee17ddf 100644 --- a/app/main.py +++ b/app/main.py @@ -28,6 +28,7 @@ oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") scanner = BackgroundScanner(db=get_db_session()) logging.basicConfig(level=logging.INFO) + @asynccontextmanager async def lifespan(app: FastAPI): verify_settings() @@ -37,17 +38,16 @@ async def lifespan(app: FastAPI): if not os.getenv("DISABLE_CARDS"): scanner.start() - logger.info("-"*63) + logger.info("-" * 63) logger.info("---- Documentation is at http://127.0.0.1:8000/api/v1/docs ----") - logger.info("-"*63) + logger.info("-" * 63) yield - #scanner.stop() + # scanner.stop() + app = FastAPI( - lifespan=lifespan, - docs_url="/api/v1/docs", - openapi_url="/api/v1/openapi.json" - ) + lifespan=lifespan, docs_url="/api/v1/docs", openapi_url="/api/v1/openapi.json" +) origins = [ "http://127.0.0.1", @@ -69,4 +69,4 @@ app.include_router(groupManager.group_router) app.include_router(cardManager.card_router) app.include_router(aaManager.aa_router) app.include_router(doorManager.door_router) -app.include_router(debugManager.debug_router) \ No newline at end of file +app.include_router(debugManager.debug_router) diff --git a/app/model/models.py b/app/model/models.py index 253887a..6935e13 100644 --- a/app/model/models.py +++ b/app/model/models.py @@ -8,89 +8,120 @@ from sqlmodel import Field, Relationship, SQLModel class Base(SQLModel): pass + #### User class UserBase(Base): name: str = Field(index=True, unique=True) email: str | None = None is_admin: bool = False + class UserResponse(UserBase): id: int + class UserCreate(UserBase): password: str + class UserDB(UserBase, table=True): id: int | None = Field(default=None, primary_key=True) passwordhash: str + class UserUpdate(Base): name: str | None = None email: str | None = None is_admin: bool | None = None password: str | None = None + #### Special class AaGroupLink(Base, table=True): - group_id: int | None = Field(default=None, foreign_key="groupdb.id", primary_key=True) - accessauth_id: int | None = Field(default=None, foreign_key="accessauthorizationdb.id", primary_key=True) + group_id: int | None = Field( + default=None, foreign_key="groupdb.id", primary_key=True + ) + accessauth_id: int | None = Field( + default=None, foreign_key="accessauthorizationdb.id", primary_key=True + ) + #### Token class Token(Base): access_token: str token_type: str + class TokenData(Base): username: str | None = None + #### Group class GroupBase(Base): name: str = Field(index=True, unique=True) + class GroupCreate(GroupBase): pass + class GroupDB(GroupBase, table=True): id: int | None = Field(default=None, primary_key=True) cards: list["Card"] = Relationship(back_populates="group") - accessauths: list["AccessAuthorizationDB"] = Relationship(back_populates="groups", link_model=AaGroupLink) + accessauths: list["AccessAuthorizationDB"] = Relationship( + back_populates="groups", link_model=AaGroupLink + ) + class GroupResponse(GroupBase): id: int cards: list["Card"] | None accessauths: list["AccessAuthorizationDB"] | None + #### AccessAuthorization class AccessAuthorizationBase(Base): name: str = Field(index=True) type: Literal["timetable", "oneshot", "somefuturespec"] is_active: bool + class AccessAuthorizationDB(AccessAuthorizationBase, table=True): id: int | None = Field(default=None, primary_key=True) type: str - groups: list["GroupDB"] = Relationship(back_populates="accessauths", link_model=AaGroupLink) - timetables: list["Timetable"] = Relationship(back_populates="accessauth", cascade_delete=True) - oneshot: "OneShotAccess" = Relationship(back_populates="accessauth", cascade_delete=True) + groups: list["GroupDB"] = Relationship( + back_populates="accessauths", link_model=AaGroupLink + ) + timetables: list["Timetable"] = Relationship( + back_populates="accessauth", cascade_delete=True + ) + oneshot: "OneShotAccess" = Relationship( + back_populates="accessauth", cascade_delete=True + ) + class OneShotAccessBase(Base): uses: int = 1 ends_at: datetime + class OneShotAccess(OneShotAccessBase, table=True): id: int | None = Field(default=None, primary_key=True) accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id") accessauth: AccessAuthorizationDB = Relationship(back_populates="oneshot") + class AccessAuthorizationCreate(AccessAuthorizationBase): timetables: list["TimetableCreate"] = [] oneshot: OneShotAccessBase | None = None - + @model_validator(mode="after") def check_type(self): if self.type == "timetable": if not self.timetables: - raise ValueError("timetable auths require at least one timetable object") + raise ValueError( + "timetable auths require at least one timetable object" + ) if self.oneshot is not None: raise ValueError("timetable auths are not allowed oneshot objects") elif self.type == "oneshot": @@ -102,19 +133,21 @@ class AccessAuthorizationCreate(AccessAuthorizationBase): raise ValueError("somefuturespec is not jet implemented. Please do not use") return self + class AccessAuthorizationResponse(AccessAuthorizationBase): id: int timetables: list["Timetable"] = [] oneshot: OneShotAccessBase | None = None groups: list["GroupDB"] + class AccessAuthorizationUpdate(Base): name: str | None = None type: Literal["timetable", "oneshot", "somefuturespec"] | None = None is_active: bool | None = None timetables: list["TimetableCreate"] | None = None oneshot: OneShotAccessBase | None = None - + #### Card class Card(Base, table=True): @@ -126,25 +159,30 @@ class Card(Base, table=True): group_id: int | None = Field(default=None, foreign_key="groupdb.id") group: GroupDB | None = Relationship(back_populates="cards") + class CardCreate(Base): name: str = Field(unique=True, max_length=32) enabled: bool = True group_id: int + class CardUpdate(Base): name: str | None = None enabled: bool | None = None group_id: int | None = None + class TimetableBase(Base): weekday: int = Field(le=6, ge=0) starttime: time duration: int = Field(gt=0, lt=1440) + class Timetable(TimetableBase, table=True): id: int | None = Field(default=None, primary_key=True) accessauth_id: int = Field(default=None, foreign_key="accessauthorizationdb.id") accessauth: AccessAuthorizationDB = Relationship(back_populates="timetables") + class TimetableCreate(TimetableBase): pass diff --git a/app/services/auth.py b/app/services/auth.py index 619c0d0..2cad156 100644 --- a/app/services/auth.py +++ b/app/services/auth.py @@ -17,7 +17,7 @@ from sqlmodel import Session, select from app.model.models import Token, TokenData, UserDB from app.services.database import * -SECRET_KEY = os.getenv("SECRET_KEY", default="ff"*16) +SECRET_KEY = os.getenv("SECRET_KEY", default="ff" * 16) ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 120 @@ -27,16 +27,20 @@ token_router = APIRouter(tags=["Token"], prefix="/api/v1") password_hash = PasswordHash.recommended() + def verify_password(plain_password, hashed_password): return password_hash.verify(plain_password, hashed_password) + def get_password_hash(password): return password_hash.hash(password) + def get_user(db, username: str): user = db.exec(select(UserDB).where(UserDB.name == username)).first() return user + def authenticate_user(db, username: str, password: str): user = get_user(db, username) if not user: @@ -45,6 +49,7 @@ def authenticate_user(db, username: str, password: str): return False return user + def create_access_token(data: dict, expires_delta: timedelta | None = None): to_encode = data.copy() if expires_delta: @@ -55,14 +60,15 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None): encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt + def get_current_user( token: Annotated[str, Depends(oauth2_scheme)], db: Session = Depends(get_session), - ): +): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", - headers={"WWW-Authenticate": "Bearer"} + headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) @@ -77,29 +83,29 @@ def get_current_user( raise credentials_exception return user + def auth_is_admin( token: str = Depends(oauth2_scheme), db: Session = Depends(get_session), - ): +): user = get_current_user(token=token, db=db) if not user.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to perform this action", - headers={"WWW-Authenticate": "Bearer"} + headers={"WWW-Authenticate": "Bearer"}, ) return True + def create_first_user(db: Session): logger.info("Checking for admin user") admin_user = db.exec(select(UserDB)).first() if admin_user is None: - password = ''.join(secrets.choice(string.digits) for i in range(8)) + password = "".join(secrets.choice(string.digits) for i in range(8)) logger.info(f"Creating first admin user with password: {password}") user = UserDB( - name="admin", - passwordhash=get_password_hash(password), - is_admin=True + name="admin", passwordhash=get_password_hash(password), is_admin=True ) return add_and_refresh(db, user) logger.info(f"Admin user already exists: {admin_user.name}") @@ -108,14 +114,14 @@ def create_first_user(db: Session): @token_router.post("/token") def login_for_access_token( form_data: Annotated[OAuth2PasswordRequestForm, Depends()], - db: Session = Depends(get_session) + db: Session = Depends(get_session), ) -> Token: user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or pw", - headers={"WWW-Authenticate": "Bearer"} + headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( @@ -123,8 +129,7 @@ def login_for_access_token( ) return Token(access_token=access_token, token_type="bearer") + @token_router.get("/test/login") -def test_login( - current_user: Annotated[UserDB, Depends(get_current_user)] -) -> UserDB: - return current_user \ No newline at end of file +def test_login(current_user: Annotated[UserDB, Depends(get_current_user)]) -> UserDB: + return current_user diff --git a/app/services/database.py b/app/services/database.py index 447fca3..9df4431 100644 --- a/app/services/database.py +++ b/app/services/database.py @@ -9,12 +9,14 @@ SQLALCHEMY_DATABASE_URL = getenv("SQLALCHEMY_DATABASE_URL", "sqlite:///./gatekee engine = create_engine(SQLALCHEMY_DATABASE_URL) + def create_db_and_tables(): if not path.exists(SQLALCHEMY_DATABASE_URL): SQLModel.metadata.create_all(engine) from alembic.config import Config from alembic import command + alembic_cfg = Config("./alembic.ini") command.stamp(alembic_cfg, "head") logger.info("Database created and tables initialized.") @@ -26,11 +28,13 @@ def get_session(): with Session(engine) as db: yield db + def get_db_session(): return Session(engine) + def add_and_refresh(db: Session, obj): db.add(obj) db.commit() db.refresh(obj) - return obj \ No newline at end of file + return obj diff --git a/app/services/door.py b/app/services/door.py index 93c508b..5b902fc 100644 --- a/app/services/door.py +++ b/app/services/door.py @@ -11,21 +11,25 @@ from app.services.database import Session, add_and_refresh doorIsOpen = True # I think this could also be gpio controlled -#See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional +# See: https://github.com/technyon/nuki_hub#gpio-lock-control-optional + def openDoor(): global doorIsOpen doorIsOpen = True logger.info("Still needs gpio out") + def closeDoor(): global doorIsOpen doorIsOpen = False logger.info("Still needs gpio out") + def isDoorOpen(): return doorIsOpen + def decrementOneshot(db: Session, oneshot: OneShotAccess): data = oneshot.model_dump() if data["uses"] > 0: @@ -33,6 +37,7 @@ def decrementOneshot(db: Session, oneshot: OneShotAccess): oneshot.sqlmodel_update(oneshot, update=data) add_and_refresh(db, oneshot) + def checkAccess(key: str, db: Session): try: current_weekday = datetime.weekday(date.today()) @@ -43,16 +48,20 @@ def checkAccess(key: str, db: Session): if auth.type == "timetable": for timetable in auth.timetables: logger.info(f" checking timetable {timetable.id}") - logger.info(f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}") + logger.info( + f" comparing weekday: CUR:{current_weekday} TT:{timetable.weekday}" + ) if current_weekday == timetable.weekday: starttime = datetime.combine(date.today(), timetable.starttime) endtime = starttime + timedelta(minutes=timetable.duration) - logger.info(f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}") + logger.info( + f" comparing time: Start:{starttime} Current:{current_time} End:{endtime}" + ) if starttime < current_time < endtime: logger.info("Access Valid!") return True if auth.type == "oneshot": - logger.info(f' oneshot auth found: {auth.oneshot}') + logger.info(f" oneshot auth found: {auth.oneshot}") if current_time < auth.oneshot.ends_at: if auth.oneshot.uses > 0: decrementOneshot(db, auth.oneshot) @@ -61,4 +70,3 @@ def checkAccess(key: str, db: Session): return False except exc.NoResultFound: raise Exception("No Access with that key found, this might be a db error") - diff --git a/app/services/scanner.py b/app/services/scanner.py index 75e1fa1..89d48bf 100644 --- a/app/services/scanner.py +++ b/app/services/scanner.py @@ -30,11 +30,11 @@ from smartcard.Exceptions import CardRequestTimeoutException from app.services.door import checkAccess, openDoor -#ENV vars +# ENV vars load_dotenv() -MIFARE_APP_MASTER_KEY = os.getenv('MIFARE_APP_MASTER_KEY') -MIFARE_ACL_READ_BASE_KEY = os.getenv('MIFARE_ACL_READ_BASE_KEY') -MIFARE_ACL_WRITE_BASE_KEY = os.getenv('MIFARE_ACL_WRITE_BASE_KEY') +MIFARE_APP_MASTER_KEY = os.getenv("MIFARE_APP_MASTER_KEY") +MIFARE_ACL_READ_BASE_KEY = os.getenv("MIFARE_ACL_READ_BASE_KEY") +MIFARE_ACL_WRITE_BASE_KEY = os.getenv("MIFARE_ACL_WRITE_BASE_KEY") # Constants MIFARE_APP_ID = "DEAFFE" # 7 bytes @@ -43,10 +43,15 @@ MIFARE_ACL_WRITE_BASE_KEY_ID = 0x2 MIFARE_SYS_ID = "FF0000" # 3 bytes, can essentially be anything MIFARE_ENCRYPTED_FILE_ID = 0x1 + def checkForKey(): if MIFARE_APP_MASTER_KEY == None: - logger.critical("NO MASTER KEY LOADED") - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="No key loaded! Check application.") + logger.critical("NO MASTER KEY LOADED") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="No key loaded! Check application.", + ) + def getCardService(timeout: int = 10): cardtype = AnyCardType() @@ -55,51 +60,58 @@ def getCardService(timeout: int = 10): cardservice.connection.connect() return cardservice + def readFileOnCard(desfire: DESFire): if not MIFARE_ACL_READ_BASE_KEY: logger.critical("MIFARE_ACL_READ_BASE_KEY not found! Reading skipped!") return - #create keys - #desfire = DESFire(PCSCDevice(cardservice.connection.component)) + # create keys + # desfire = DESFire(PCSCDevice(cardservice.connection.component)) aes_keysettings = KeySettings(key_type=DESFireKeyType.DF_KEY_AES) keysettings = desfire.get_key_setting() desKey = DESFireKey(keysettings, "00" * 8) # Get real UID desfire.authenticate(0x0, desKey) - #To get the uid you have to auth with an empty (default) key + # To get the uid you have to auth with an empty (default) key uid = desfire.get_real_uid() applications = desfire.get_application_ids() try: assert len(applications) == 1 - assert applications[0] == get_list(MIFARE_APP_ID) + assert applications[0] == get_list(MIFARE_APP_ID) except AssertionError: logger.error("No application found!") time.sleep(4) return - #Then use the key derivation with that uid, the appid, the sysid - diversification_data = [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID) - read_div_key_bytes = diversify_key(get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False) - - #Log in with derived read key + # Then use the key derivation with that uid, the appid, the sysid + diversification_data = ( + [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID) + ) + read_div_key_bytes = diversify_key( + get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False + ) + + # Log in with derived read key logger.debug("Start auth") aes_app_read_key = DESFireKey(aes_keysettings, read_div_key_bytes) desfire.select_application(MIFARE_APP_ID) desfire.authenticate(MIFARE_ACL_READ_BASE_KEY_ID, aes_app_read_key) - + logger.debug(f"Read data from {MIFARE_ENCRYPTED_FILE_ID}") file_data = desfire.get_file_settings(MIFARE_ENCRYPTED_FILE_ID) rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data) - #convert list of int to str + # convert list of int to str rdata = to_hex_string(rdata).replace(" ", "").lower() logger.debug(f"Data on card: {rdata}") return rdata + def DeleteCard(): try: checkForKey() from app.main import scanner as scannerThread + scannerThread.stop() cardservice = getCardService(15) @@ -114,7 +126,7 @@ def DeleteCard(): desKey = DESFireKey(des_keysettings, "00" * 8) aes_master_key = DESFireKey(aes_keysettings, MIFARE_APP_MASTER_KEY) aes_null_key = DESFireKey(aes_keysettings, "00" * 16) - + desfire.select_application(0x0) try: @@ -127,11 +139,13 @@ def DeleteCard(): except: logger.debug("Auth3") desfire.authenticate(0x0, desKey) - + applications = desfire.get_application_ids() logger.debug(f"Applications: {applications}") if len(applications) == 0: - raise HTTPException(status_code=status.HTTP_410_GONE, detail="No applications on card") + raise HTTPException( + status_code=status.HTTP_410_GONE, detail="No applications on card" + ) desfire.select_application(MIFARE_APP_ID) desfire.authenticate(0x0, aes_master_key) @@ -143,14 +157,18 @@ def DeleteCard(): pass scannerThread.start() return rdata - except(Exception, AssertionError) as e: + except (Exception, AssertionError) as e: logger.error(f"Error in deletion function: {e}", exc_info=True) - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}" + ) + def WriteNewCard(): try: checkForKey() from app.main import scanner as scannerThread + scannerThread.stop() cardservice = getCardService(20) @@ -166,8 +184,8 @@ def WriteNewCard(): # Authenticate with default DES key logger.debug("Authenticating with default DES key...") desfire.authenticate(0x0, desKey) - - #get uid + + # get uid uid = desfire.get_real_uid() # Set default key @@ -196,7 +214,7 @@ def WriteNewCard(): # Select application desfire.select_application(MIFARE_APP_ID) - #recreate key object + # recreate key object desfire.authenticate(0x0, aes_null_key) desfire.change_key(0x0, aes_null_key, aes_master_key, 0x1) @@ -205,18 +223,28 @@ def WriteNewCard(): aes_null_key = DESFireKey(aes_keysettings, "00" * 16) - #generate div data - diversification_data = [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID) - read_div_key_bytes = diversify_key(get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False) - write_div_key_bytes = diversify_key(get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False) - + # generate div data + diversification_data = ( + [0x01] + uid + get_list(MIFARE_APP_ID) + get_list(MIFARE_SYS_ID) + ) + read_div_key_bytes = diversify_key( + get_list(MIFARE_ACL_READ_BASE_KEY), diversification_data, pad_to_32=False + ) + write_div_key_bytes = diversify_key( + get_list(MIFARE_ACL_WRITE_BASE_KEY), diversification_data, pad_to_32=False + ) + logger.debug("Changing file read key...") aes_file_read_key = DESFireKey(aes_keysettings, read_div_key_bytes) - desfire.change_key(MIFARE_ACL_READ_BASE_KEY_ID, aes_null_key, aes_file_read_key, 0x1) + desfire.change_key( + MIFARE_ACL_READ_BASE_KEY_ID, aes_null_key, aes_file_read_key, 0x1 + ) logger.debug("Changing file write key...") aes_file_write_key = DESFireKey(aes_keysettings, write_div_key_bytes) - desfire.change_key(MIFARE_ACL_WRITE_BASE_KEY_ID, aes_null_key, aes_file_write_key, 0x1) + desfire.change_key( + MIFARE_ACL_WRITE_BASE_KEY_ID, aes_null_key, aes_file_write_key, 0x1 + ) logger.debug("Create encrypted file containing key...") file_settings = FileSettings( @@ -233,18 +261,22 @@ def WriteNewCard(): logger.debug("Writing UID to encrypted file...") key = secrets.token_hex(16) - desfire.write_file_data(MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(key)) + desfire.write_file_data( + MIFARE_ENCRYPTED_FILE_ID, 0x0, file_data.encryption, get_list(key) + ) logger.debug("Reading from encrypted file...") rdata = desfire.read_file_data(MIFARE_ENCRYPTED_FILE_ID, file_data) assert rdata == get_list(key) logger.debug(" - Data written successfully.") scannerThread.start() - return key, to_hex_string(data=uid, separator=":") + return key, to_hex_string(data=uid, separator=":") except Exception as e: logger.error(f"Error in write function: {e}", exc_info=True) - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error: {e}" + ) class BackgroundScanner: @@ -299,7 +331,7 @@ class BackgroundScanner: except Exception as e: logger.error(f"something went wrong: {e}") time.sleep(5) - + def _check_db(self, key): check = checkAccess(key, self.db) if check == True: diff --git a/app/services/settings.py b/app/services/settings.py index 2d42c98..4a5a2bf 100644 --- a/app/services/settings.py +++ b/app/services/settings.py @@ -3,25 +3,27 @@ import os logger = logging.getLogger(__name__) + def verify_settings(): - card_envs = [ + card_envs = [ "MIFARE_APP_MASTER_KEY", "MIFARE_ACL_READ_BASE_KEY", "MIFARE_ACL_WRITE_BASE_KEY", ] - important_envs = [ - "SECRET_KEY" - ] - other_envs = [ - "SQLALCHEMY_DATABASE_URL" - ] + important_envs = ["SECRET_KEY"] + other_envs = ["SQLALCHEMY_DATABASE_URL"] for setting in card_envs: - if (setting not in os.environ or setting == "") and not os.getenv("DISABLE_CARDS"): - raise ValueError(f"Missing environment variable for scanner start: {setting} \n Run with DISABLE_CARDS env var to disable cards") + if (setting not in os.environ or setting == "") and not os.getenv( + "DISABLE_CARDS" + ): + raise ValueError( + f"Missing environment variable for scanner start: {setting} \n Run with DISABLE_CARDS env var to disable cards" + ) for setting in important_envs: if setting not in os.environ or setting == "": - raise ValueError(f'Missing critical environment variable {setting}. Stopping...') + raise ValueError( + f"Missing critical environment variable {setting}. Stopping..." + ) for setting in other_envs: if setting not in os.environ: - logger.critical(f'Env var {setting} not set. Continuing with defaults.') - + logger.critical(f"Env var {setting} not set. Continuing with defaults.")