Add endpoint for creating local contacts with access control

Introduced a new `local_contact_router` to handle creation of local contacts. The endpoint enforces role-based access control and ensures no duplication of email addresses. Updated the router exports for consistency and cleaned up a large test file to improve readability.
This commit is contained in:
GotthardG
2025-02-26 09:58:19 +01:00
parent 43d67b1044
commit f588bc0cda
13 changed files with 360 additions and 418 deletions

View File

@@ -21,6 +21,12 @@ mock_users_db = {
"password": "testpass2", # In a real scenario, store the hash of the password
"pgroups": ["p20004", "p20005", "p20006"],
},
"admin": {
"username": "admin",
"password": "adminpass",
"pgroups": ["p20007"],
# "role": "admin",
},
}
@@ -49,6 +55,9 @@ async def get_current_user(token: str = Depends(oauth2_scheme)) -> loginData:
username: str = payload.get("sub")
print(f"[DEBUG] Username decoded from token: {username}") # Add debug log here
return loginData(username=username, pgroups=payload.get("pgroups"))
# return loginData(username=username, pgroups=payload.get("pgroups"),
# role=payload.get("role"))
except jwt.ExpiredSignatureError:
print("[DEBUG] Token expired")
raise HTTPException(status_code=401, detail="Token expired")
@@ -57,6 +66,14 @@ async def get_current_user(token: str = Depends(oauth2_scheme)) -> loginData:
raise HTTPException(status_code=401, detail="Invalid token")
# async def get_user_role(token: str = Depends(oauth2_scheme)) -> str:
# try:
# payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
# return payload.get("role")
# except jwt.ExpiredSignatureError:
# raise HTTPException(status_code=401, detail="Token expired")
@router.post("/token/login", response_model=loginToken)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = mock_users_db.get(form_data.username)
@@ -70,10 +87,14 @@ async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# Create token
access_token = create_access_token(
data={"sub": user["username"], "pgroups": user["pgroups"]}
# data = {"sub": user["username"], "pgroups": user["pgroups"],
# "role": user["role"]}
)
return loginToken(access_token=access_token, token_type="bearer")
@router.get("/protected-route")
async def read_protected_data(current_user: loginData = Depends(get_current_user)):
# return {"username": current_user.username, "pgroups":
# current_user.pgroups, "role": current_user.role}
return {"username": current_user.username, "pgroups": current_user.pgroups}