Add Linkki broadcast image

linkki-tiedotus/webapp/auth.py

@@ -0,0 +1,169 @@
+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()