122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
import datetime as dt
|
|
import grp
|
|
import pwd
|
|
from typing import List, Set, Optional
|
|
|
|
import jwt
|
|
import pam
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
|
|
from backend.settings import Settings, get_settings
|
|
|
|
|
|
def _is_member_of(username: str, groups: Set[str]) -> bool:
|
|
if not groups:
|
|
return False
|
|
try:
|
|
user_entry = pwd.getpwnam(username)
|
|
except KeyError:
|
|
return False
|
|
user_groups = {g.gr_name for g in grp.getgrall() if username in g.gr_mem}
|
|
primary_group = grp.getgrgid(user_entry.pw_gid).gr_name
|
|
user_groups.add(primary_group)
|
|
return bool(user_groups & groups)
|
|
|
|
|
|
def is_authorized_admin(username: str, settings: Settings, mode: Optional[str] = None) -> bool:
|
|
if mode == "oidc":
|
|
allowed_users = set(settings.auth_allowed_users)
|
|
if allowed_users and username not in allowed_users:
|
|
return False
|
|
return True
|
|
# UID 0 always allowed
|
|
try:
|
|
entry = pwd.getpwnam(username)
|
|
except KeyError:
|
|
return False
|
|
if entry.pw_uid == 0:
|
|
return True
|
|
allowed_users = set(settings.auth_allowed_users)
|
|
allowed_groups = set(settings.auth_allowed_groups)
|
|
if username in allowed_users:
|
|
return True
|
|
if _is_member_of(username, allowed_groups):
|
|
return True
|
|
return False
|
|
|
|
|
|
def authenticate_admin_user(username: str, password: str, settings: Settings) -> None:
|
|
if not is_authorized_admin(username, settings, mode="pam"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="User not authorized to log in",
|
|
)
|
|
pam_client = pam.pam()
|
|
if not pam_client.authenticate(username, password, service=settings.auth_pam_service):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid credentials",
|
|
)
|
|
|
|
|
|
def issue_token(username: str, settings: Settings) -> str:
|
|
payload = {
|
|
"sub": username,
|
|
"exp": dt.datetime.utcnow() + dt.timedelta(seconds=settings.token_ttl_seconds),
|
|
"iat": dt.datetime.utcnow(),
|
|
}
|
|
return jwt.encode(payload, settings.auth_secret, algorithm="HS256")
|
|
|
|
|
|
def decode_token(token: str, settings: Settings) -> str:
|
|
try:
|
|
payload = jwt.decode(token, settings.auth_secret, algorithms=["HS256"])
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired") from exc
|
|
except jwt.PyJWTError as exc:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") from exc
|
|
|
|
username = payload.get("sub")
|
|
if not username:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
if not (
|
|
is_authorized_admin(username, settings, mode="pam")
|
|
or is_authorized_admin(username, settings, mode="oidc")
|
|
):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Unauthorized user")
|
|
return username
|
|
|
|
|
|
def get_current_admin(
|
|
request: Request,
|
|
settings: Settings = Depends(get_settings),
|
|
) -> str:
|
|
auth_header = request.headers.get("authorization")
|
|
token = None
|
|
if auth_header and auth_header.lower().startswith("bearer "):
|
|
token = auth_header.split(" ", 1)[1].strip()
|
|
if not token:
|
|
token = request.cookies.get(settings.session_cookie_name)
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing token")
|
|
return decode_token(token, settings)
|
|
|
|
|
|
def list_manageable_users(settings: Settings) -> List[str]:
|
|
"""List system users we are allowed to manage (uid >= 1000, real shells, optional allowlist)."""
|
|
allowed = set(settings.allowed_users) if settings.allowed_users else None
|
|
candidates: List[str] = []
|
|
for entry in pwd.getpwall():
|
|
if entry.pw_uid == 0:
|
|
# Never operate on root accounts
|
|
continue
|
|
if entry.pw_uid < 1000:
|
|
continue
|
|
if entry.pw_shell in ("/usr/sbin/nologin", "/bin/false"):
|
|
continue
|
|
if allowed is not None and entry.pw_name not in allowed:
|
|
continue
|
|
candidates.append(entry.pw_name)
|
|
candidates.sort()
|
|
return candidates
|