Plex-OAuth/app.py at main · Kometa-Team/Plex-OAuth · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
"""
Plex OAuth Flask Application
A minimal Flask app for Plex OAuth authentication
"""

import os
import uuid

import requests
from flask import Flask, jsonify, redirect, render_template, request, session, url_for

app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY", os.urandom(24))

# Plex API endpoints
PLEX_API_URL = "https://plex.tv/api/v2"
PLEX_AUTH_URL = "https://app.plex.tv/auth"

# App configuration
APP_NAME = os.environ.get("APP_NAME", "Kometa Plex Auth")
APP_VERSION = os.environ.get("APP_VERSION", "1.0")


def get_client_identifier():
    """Get or create a client identifier for this session."""
    if "client_id" not in session:
        session["client_id"] = str(uuid.uuid4())
    return session["client_id"]


def get_plex_headers():
    """Get common headers for Plex API requests."""
    return {
        "X-Plex-Client-Identifier": get_client_identifier(),
        "X-Plex-Product": APP_NAME,
        "X-Plex-Version": APP_VERSION,
        "Accept": "application/json",
    }


@app.route("/")
def index():
    """Home page with authentication button."""
    return render_template("index.html", app_name=APP_NAME)


@app.route("/auth/start")
def auth_start():
    """Start the authentication flow by creating a PIN."""
    try:
        # Create a PIN
        response = requests.post(
            f"{PLEX_API_URL}/pins",
            headers=get_plex_headers(),
            params={"strong": "true"},
        )
        response.raise_for_status()
        pin_data = response.json()

        # Store PIN ID in session
        session["pin_id"] = pin_data["id"]
        session["pin_code"] = pin_data["code"]

        # Construct auth URL
        base_url = request.url_root.rstrip("/")
        forward_url = f"{base_url}/auth/callback"

        auth_url = (
            f"{PLEX_AUTH_URL}#?"
            f"clientID={get_client_identifier()}&"
            f"code={pin_data['code']}&"
            f"forwardUrl={forward_url}&"
            f"context[device][product]={APP_NAME}"
        )

        # Redirect to Plex auth
        return redirect(auth_url)

    except requests.RequestException as e:
        return render_template(
            "error.html", error=f"Failed to start authentication: {str(e)}"
        )


@app.route("/auth/callback")
def auth_callback():
    """Handle the callback from Plex after authentication."""
    # This page just closes the popup and notifies the opener
    return """
    <!DOCTYPE html>
    <html>
    <head>
        <title>Authentication Complete</title>
    </head>
    <body>
        <script>
            if (window.opener) {
                window.opener.postMessage({type: 'plex-auth-complete'}, '*');
                window.close();
            } else {
                document.body.innerHTML = '<p>Authentication complete. You can close this window.</p>';
            }
        </script>
    </body>
    </html>
    """


@app.route("/auth/check")
def auth_check():
    """Check the status of the PIN and return auth token if available."""
    pin_id = session.get("pin_id")

    if not pin_id:
        return jsonify({"error": "No authentication session found"}), 400

    try:
        # Check the PIN to get the auth token
        response = requests.get(
            f"{PLEX_API_URL}/pins/{pin_id}", headers=get_plex_headers()
        )
        response.raise_for_status()
        pin_data = response.json()

        auth_token = pin_data.get("authToken")

        if auth_token:
            # Get user info
            user_response = requests.get(
                f"{PLEX_API_URL}/user",
                headers={**get_plex_headers(), "X-Plex-Token": auth_token},
            )
            user_response.raise_for_status()
            user_data = user_response.json()

            return jsonify(
                {
                    "authenticated": True,
                    "token": auth_token,
                    "username": user_data.get("username", "Unknown"),
                    "email": user_data.get("email", "Unknown"),
                }
            )
        else:
            return jsonify({"authenticated": False})

    except requests.RequestException as e:
        return jsonify({"error": f"Failed to verify authentication: {str(e)}"}), 500


@app.route("/health")
def health():
    """Health check endpoint for monitoring."""
    return jsonify({"status": "healthy"})


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    app.run(host="0.0.0.0", port=port, debug=os.environ.get("DEBUG", "False") == "True")