docker-images/linkki-tiedotus/webapp/auth.py

import base64
import getpass
import hashlib
import hmac
import json
import os
from pathlib import Path

from webapp.config import ConfigFileError


PBKDF2_ALGORITHM = "sha256"
PBKDF2_ITERATIONS = 600000
SALT_BYTES = 16
SUPPORTED_TRANSLATION_LANGUAGES = ["en", "fi", "sv"]
SUPPORTED_UI_LANGUAGES = ["auto", "fi", "sv", "en"]
PERMISSION_KEYS = [
    "edit_content",
    "edit_targets",
    "save_announcements",
    "open_telegram",
    "post_now",
    "schedule",
    "unschedule",
    "unlink_telegram",
    "delete_telegram",
    "delete_storage",
]


def load_users(users_file: Path) -> dict[str, dict]:
    if not users_file.exists():
        return {}

    try:
        payload = json.loads(users_file.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise ConfigFileError(users_file, f"Invalid JSON: {error.msg}") from error
    users = payload.get("users", [])
    result: dict[str, dict] = {}
    changed = False
    for user in users:
        username = str(user.get("username", "")).strip()
        password_hash = str(user.get("password_hash", "")).strip()
        if username and password_hash:
            permissions_payload, permissions_changed = normalize_permissions_payload(user.get("permissions", {}))
            if permissions_changed:
                user["permissions"] = permissions_payload
                changed = True

            language = normalize_ui_language(user.get("language", "en"))
            if user.get("language") != language:
                user["language"] = language
                changed = True

            result[username] = {
                "password_hash": password_hash,
                "permissions": load_permissions(permissions_payload),
                "language": language,
            }
    if changed:
        users_file.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")
    return result


def load_permissions(payload: object) -> dict[str, bool]:
    source = payload if isinstance(payload, dict) else {}
    permissions: dict[str, bool] = {}
    for key in PERMISSION_KEYS:
        permissions[key] = as_permission_bool(source.get(key, False))
    return permissions


def normalize_permissions_payload(payload: object) -> tuple[dict[str, object], bool]:
    source = dict(payload) if isinstance(payload, dict) else {}
    changed = not isinstance(payload, dict)
    normalized = {key: source[key] for key in source if key in PERMISSION_KEYS}
    if len(normalized) != len(source):
        changed = True
    for key in PERMISSION_KEYS:
        normalized_value = as_permission_bool(source.get(key, False))
        if source.get(key) != normalized_value:
            changed = True
        normalized[key] = normalized_value
    return normalized, changed


def as_permission_bool(value: object) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, int):
        return value != 0
    if isinstance(value, str):
        lowered = value.strip().lower()
        return lowered in {"1", "true", "yes", "on"}
    return False


def normalize_ui_language(value: object) -> str:
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in SUPPORTED_UI_LANGUAGES:
            return lowered
    return "en"


def normalize_translation_language(value: object) -> str:
    if isinstance(value, str):
        lowered = value.strip().lower()
        if lowered in SUPPORTED_TRANSLATION_LANGUAGES:
            return lowered
    return "en"


def set_user_language(users_file: Path, username: str, language: str) -> None:
    normalized = normalize_ui_language(language)
    try:
        payload = json.loads(users_file.read_text(encoding="utf-8"))
    except json.JSONDecodeError as error:
        raise ConfigFileError(users_file, f"Invalid JSON: {error.msg}") from error
    users = payload.get("users", [])
    updated = False
    for user in users:
        if str(user.get("username", "")).strip() == username:
            user["language"] = normalized
            updated = True
            break

    if not updated:
        raise ValueError(f"Unknown user: {username}")

    users_file.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8")


def hash_password(password: str) -> str:
    salt = os.urandom(SALT_BYTES)
    digest = hashlib.pbkdf2_hmac(PBKDF2_ALGORITHM, password.encode("utf-8"), salt, PBKDF2_ITERATIONS)
    return "$".join(
        [
            "pbkdf2_sha256",
            str(PBKDF2_ITERATIONS),
            base64.urlsafe_b64encode(salt).decode("ascii"),
            base64.urlsafe_b64encode(digest).decode("ascii"),
        ]
    )


def verify_password(password: str, password_hash: str) -> bool:
    algorithm, iterations_text, salt_text, digest_text = password_hash.split("$", 3)
    if algorithm != "pbkdf2_sha256":
        return False

    iterations = int(iterations_text)
    salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
    expected_digest = base64.urlsafe_b64decode(digest_text.encode("ascii"))
    actual_digest = hashlib.pbkdf2_hmac(PBKDF2_ALGORITHM, password.encode("utf-8"), salt, iterations)
    return hmac.compare_digest(actual_digest, expected_digest)


def hash_password_cli() -> None:
    password = getpass.getpass("Password: ")
    password_repeat = getpass.getpass("Repeat password: ")
    if password != password_repeat:
        raise ValueError("Passwords did not match.")
    print(hash_password(password))


if __name__ == "__main__":
    hash_password_cli()