computer-playbook/roles/web-app-jellyfin/files/python/provision_mcp_identity.py

"""Give the MCP adapter a read-only Jellyfin user and print its access token.

Jellyfin mints every token itself, so the deployment cannot pin one. What it
can pin is the user: the password is a deployment credential, the policy is
written explicitly, and the token is captured from the authentication that
follows.

An API key is not an option here. Jellyfin resolves every API key to the
administrator role, so the least-privileged identity that still reads the
contracted endpoints is a user token.

Prints the access token on stdout, prefixed by CHANGED or UNCHANGED.

Environment:
    JELLYFIN_BASE:           server base URL.
    JELLYFIN_ADMIN_USER:     administrator to act as.
    JELLYFIN_ADMIN_PASSWORD: that administrator's password.
    MCP_USER:                username to converge.
    MCP_PASSWORD:            password to pin on it.
    MCP_DEVICE_ID:           stable device id, so re-runs reuse one session.
    MCP_CURRENT_TOKEN:       token the adapter holds today, or "" for none.
"""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request

BASE = os.environ["JELLYFIN_BASE"].rstrip("/")
ADMIN_USER = os.environ["JELLYFIN_ADMIN_USER"]
ADMIN_PASSWORD = os.environ["JELLYFIN_ADMIN_PASSWORD"]
MCP_USER = os.environ["MCP_USER"]
MCP_PASSWORD = os.environ["MCP_PASSWORD"]
DEVICE_ID = os.environ["MCP_DEVICE_ID"]
CURRENT_TOKEN = os.environ.get("MCP_CURRENT_TOKEN", "").strip()

TIMEOUT = 30
CLIENT = (
    'MediaBrowser Client="infinito", Device="ansible", '
    f'DeviceId="{DEVICE_ID}", Version="1.0.0"'
)

READ_ONLY_POLICY = {
    "IsAdministrator": False,
    "IsHidden": True,
    "IsDisabled": False,
    "EnableAllFolders": True,
    "EnableRemoteAccess": True,
    "EnableMediaPlayback": False,
    "EnableContentDeletion": False,
    "EnableContentDownloading": False,
    "EnableCollectionManagement": False,
    "EnableSubtitleManagement": False,
    "EnableLyricManagement": False,
    "EnableLiveTvManagement": False,
    "EnableLiveTvAccess": False,
    "EnableRemoteControlOfOtherUsers": False,
    "EnableSharedDeviceControl": False,
    "EnableMediaConversion": False,
    "EnableAudioPlaybackTranscoding": False,
    "EnableVideoPlaybackTranscoding": False,
    "EnablePlaybackRemuxing": False,
    "EnablePublicSharing": False,
    "EnableSyncTranscoding": False,
    "EnableUserPreferenceAccess": False,
    "AuthenticationProviderId": (
        "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider"
    ),
    "PasswordResetProviderId": (
        "Jellyfin.Server.Implementations.Users.DefaultPasswordResetProvider"
    ),
}


TRANSIENT_STATUS = frozenset({0, 429, 502, 503, 504})


def fail(message, status=None):
    """Abort the run, marking only a terminal cause as REJECTED.

    Args:
        message: what went wrong.
        status: the HTTP status that caused it, when there was one.
    """
    marker = "PENDING" if status in TRANSIENT_STATUS else "REJECTED"
    sys.stderr.write(f"{marker} {message}\n")
    sys.exit(1)


def call(method, path, payload=None, token=None):
    """Return ``(status, parsed body)`` of one Jellyfin API call.

    Args:
        method: HTTP method.
        path: path below the server base.
        payload: JSON body, or None.
        token: access token to authenticate with, or None for the client header.
    """
    data = json.dumps(payload).encode() if payload is not None else None
    request = urllib.request.Request(f"{BASE}{path}", data=data, method=method)  # noqa: S310 fixed internal http origin
    request.add_header("Accept", "application/json")
    request.add_header(
        "Authorization", f'{CLIENT}, Token="{token}"' if token else CLIENT
    )
    if data is not None:
        request.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT) as response:  # noqa: S310 fixed internal http origin
            body = response.read()
            return response.status, json.loads(body) if body else None
    except urllib.error.HTTPError as error:
        return error.code, error.read().decode(errors="replace")
    except OSError as error:
        return 0, f"unreachable: {error}"


def admin_token():
    status, body = call(
        "POST",
        "/Users/AuthenticateByName",
        {"Username": ADMIN_USER, "Pw": ADMIN_PASSWORD},
    )
    if status != 200 or not isinstance(body, dict):
        fail(
            f"authenticating {ADMIN_USER} answered {status}: {str(body)[:200]}", status
        )
    return body["AccessToken"]


def ensure_user(token):
    status, body = call("GET", "/Users", token=token)
    if status != 200:
        fail(f"listing users answered {status}: {str(body)[:200]}", status)

    for user in body or []:
        if user.get("Name") == MCP_USER:
            return user["Id"], False

    status, body = call(
        "POST", "/Users/New", {"Name": MCP_USER, "Password": MCP_PASSWORD}, token=token
    )
    if status not in (200, 204) or not isinstance(body, dict):
        fail(f"creating {MCP_USER} answered {status}: {str(body)[:200]}", status)
    return body["Id"], True


def set_password(token, user_id):
    """Converge the inventory password onto a user that already exists.

    Args:
        token: an administrator access token; the server skips the
            current-password check only for an admin acting on another user.
        user_id: the target user's id, passed as a query parameter because
            omitting it retargets the caller's own account.
    """
    status, body = call(
        "POST",
        f"/Users/Password?userId={user_id}",
        {"NewPw": MCP_PASSWORD, "ResetPassword": False},
        token=token,
    )
    if status not in (200, 204):
        fail(
            f"setting the password of {MCP_USER} answered {status}: {str(body)[:200]}",
            status,
        )


def main():
    status, body = call(
        "POST", "/Users/AuthenticateByName", {"Username": MCP_USER, "Pw": MCP_PASSWORD}
    )
    if status == 200 and isinstance(body, dict):
        if CURRENT_TOKEN:
            probe, _ = call("GET", "/System/Info", token=CURRENT_TOKEN)
            if probe == 200:
                print("UNCHANGED")
                print(CURRENT_TOKEN)
                return
        print("CHANGED")
        print(body["AccessToken"])
        return

    token = admin_token()
    user_id, _created = ensure_user(token)

    status, body = call(
        "POST", f"/Users/{user_id}/Policy", READ_ONLY_POLICY, token=token
    )
    if status not in (200, 204):
        fail(f"restricting {MCP_USER} answered {status}: {str(body)[:200]}", status)

    set_password(token, user_id)

    status, body = call(
        "POST",
        "/Users/AuthenticateByName",
        {"Username": MCP_USER, "Pw": MCP_PASSWORD},
    )
    if status != 200 or not isinstance(body, dict):
        fail(f"authenticating {MCP_USER} answered {status}: {str(body)[:200]}", status)

    print("CHANGED")
    print(body["AccessToken"])


if __name__ == "__main__":
    main()