jellyfin-accounts/jellyfin_accounts/web_api.py

331 lines
11 KiB
Python

from flask import request, jsonify
from jellyfin_accounts.jf_api import Jellyfin
import json
import datetime
import secrets
import time
from jellyfin_accounts import config, config_path, app, g, data_store
from jellyfin_accounts import web_log as log
from jellyfin_accounts.validate_password import PasswordValidator
def resp(success=True, code=500):
if success:
r = jsonify({"success": True})
if code == 500:
r.status_code = 200
else:
r.status_code = code
else:
r = jsonify({"success": False})
r.status_code = code
return r
def checkInvite(code, delete=False):
current_time = datetime.datetime.now()
invites = dict(data_store.invites)
match = False
for invite in invites:
expiry = datetime.datetime.strptime(
invites[invite]["valid_till"], "%Y-%m-%dT%H:%M:%S.%f"
)
if current_time >= expiry:
log.debug(f"Housekeeping: Deleting old invite {invite}")
del data_store.invites[invite]
elif invite == code:
match = True
if delete:
del data_store.invites[code]
return match
jf = Jellyfin(
config["jellyfin"]["server"],
config["jellyfin"]["client"],
config["jellyfin"]["version"],
config["jellyfin"]["device"],
config["jellyfin"]["device_id"],
)
from jellyfin_accounts.login import auth
jf_address = config["jellyfin"]["server"]
success = False
for i in range(3):
try:
jf.authenticate(config["jellyfin"]["username"], config["jellyfin"]["password"])
success = True
log.info(f"Successfully authenticated with {jf_address}")
break
except Jellyfin.AuthenticationError:
log.error(f"Failed to authenticate with {jf_address}, Retrying...")
time.sleep(5)
if not success:
log.error("Could not authenticate after 3 tries.")
exit()
# Temporary fixes below.
def switchToIds():
try:
with open(config["files"]["emails"], "r") as f:
emails = json.load(f)
except (FileNotFoundError, json.decoder.JSONDecodeError):
emails = {}
users = jf.getUsers(public=False)
new_emails = {}
match = False
for key in emails:
for user in users:
if user["Name"] == key:
match = True
new_emails[user["Id"]] = emails[key]
elif user["Id"] == key:
new_emails[user["Id"]] = emails[key]
if match:
from pathlib import Path
email_file = Path(config["files"]["emails"]).name
log.info(
(
f"{email_file} modified to use userID instead of "
+ "usernames. These will be used in future."
)
)
emails = new_emails
with open(config["files"]["emails"], "w") as f:
f.write(json.dumps(emails, indent=4))
# Temporary, switches emails.json over from using Usernames to User IDs.
switchToIds()
from packaging import version
if (
version.parse(jf.info["Version"]) >= version.parse("10.6.0")
and bool(data_store.user_template) is not False
):
log.info("Updating user_template for Jellyfin >= 10.6.0")
if (
data_store.user_template["AuthenticationProviderId"]
== "Emby.Server.Implementations.Library.DefaultAuthenticationProvider"
):
data_store.user_template[
"AuthenticationProviderId"
] = "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider"
if (
data_store.user_template["PasswordResetProviderId"]
== "Emby.Server.Implementations.Library.DefaultPasswordResetProvider"
):
data_store.user_template[
"PasswordResetProviderId"
] = "Jellyfin.Server.Implementations.Users.DefaultPasswordResetProvider"
if config.getboolean("password_validation", "enabled"):
validator = PasswordValidator(
config["password_validation"]["min_length"],
config["password_validation"]["upper"],
config["password_validation"]["lower"],
config["password_validation"]["number"],
config["password_validation"]["special"],
)
else:
validator = PasswordValidator(0, 0, 0, 0, 0)
@app.route("/newUser", methods=["POST"])
def newUser():
data = request.get_json()
log.debug("Attempted newUser")
if checkInvite(data["code"]):
validation = validator.validate(data["password"])
valid = True
for criterion in validation:
if validation[criterion] is False:
valid = False
if valid:
log.debug("User password valid")
try:
user = jf.newUser(data["username"], data["password"])
except Jellyfin.UserExistsError:
error = f'User already exists named {data["username"]}'
log.debug(error)
return jsonify({"error": error})
except:
return jsonify({"error": "Unknown error"})
checkInvite(data["code"], delete=True)
if user.status_code == 200:
try:
policy = data_store.user_template
if policy != {}:
jf.setPolicy(user.json()["Id"], policy)
else:
log.debug("user policy was blank")
except:
log.error("Failed to set new user policy")
try:
configuration = data_store.user_configuration
displayprefs = data_store.user_displayprefs
if configuration != {} and displayprefs != {}:
if jf.setConfiguration(user.json()["Id"], configuration):
jf.setDisplayPreferences(user.json()["Id"], displayprefs)
log.debug("Set homescreen layout.")
else:
log.debug(
"user configuration and/or " + "displayprefs were blank"
)
except:
log.error("Failed to set new user homescreen layout")
if config.getboolean("password_resets", "enabled"):
data_store.emails[user.json()["Id"]] = data["email"]
log.debug("Email address stored")
log.info("New user created")
else:
log.error(f"New user creation failed: {user.status_code}")
return resp(False)
else:
log.debug("User password invalid")
return jsonify(validation)
else:
log.debug("Attempted newUser unauthorized")
return resp(False, code=401)
@app.route("/generateInvite", methods=["POST"])
@auth.login_required
def generateInvite():
current_time = datetime.datetime.now()
data = request.get_json()
delta = datetime.timedelta(hours=int(data["hours"]), minutes=int(data["minutes"]))
invite_code = secrets.token_urlsafe(16)
invite = {}
log.debug(f"Creating new invite: {invite_code}")
valid_till = current_time + delta
invite["valid_till"] = valid_till.strftime("%Y-%m-%dT%H:%M:%S.%f")
if "email" in data and config.getboolean("invite_emails", "enabled"):
address = data["email"]
invite["email"] = address
log.info(f"Sending invite to {address}")
method = config["email"]["method"]
if method == "mailgun":
from jellyfin_accounts.email import Mailgun
email = Mailgun(address)
elif method == "smtp":
from jellyfin_accounts.email import Smtp
email = Smtp(address)
email.construct_invite({"expiry": valid_till, "code": invite_code})
response = email.send()
if response is False or type(response) != bool:
invite["email"] = f"Failed to send to {address}"
data_store.invites[invite_code] = invite
log.info(f"New invite created: {invite_code}")
return resp()
@app.route("/getInvites", methods=["GET"])
@auth.login_required
def getInvites():
log.debug("Invites requested")
current_time = datetime.datetime.now()
invites = dict(data_store.invites)
for code in invites:
checkInvite(code)
invites = dict(data_store.invites)
response = {"invites": []}
for code in invites:
expiry = datetime.datetime.strptime(
invites[code]["valid_till"], "%Y-%m-%dT%H:%M:%S.%f"
)
valid_for = expiry - current_time
invite = {
"code": code,
"hours": valid_for.seconds // 3600,
"minutes": (valid_for.seconds // 60) % 60,
}
if "email" in invites[code]:
invite["email"] = invites[code]["email"]
response["invites"].append(invite)
return jsonify(response)
@app.route("/deleteInvite", methods=["POST"])
@auth.login_required
def deleteInvite():
code = request.get_json()["code"]
invites = dict(data_store.invites)
if code in invites:
del data_store.invites[code]
log.info(f"Invite deleted: {code}")
return resp()
@app.route("/getToken")
@auth.login_required
def get_token():
token = g.user.generate_token()
return jsonify({"token": token.decode("ascii")})
@app.route("/getUsers", methods=["GET"])
@auth.login_required
def getUsers():
log.debug("User and email list requested")
response = {"users": []}
users = jf.getUsers(public=False)
emails = data_store.emails
for user in users:
entry = {"name": user["Name"]}
if user["Id"] in emails:
entry["email"] = emails[user["Id"]]
response["users"].append(entry)
return jsonify(response)
@app.route("/modifyUsers", methods=["POST"])
@auth.login_required
def modifyUsers():
data = request.get_json()
log.debug("Email list modification requested")
for key in data:
uid = jf.getUsers(key, public=False)["Id"]
data_store.emails[uid] = data[key]
log.debug(f'Email for user "{key}" modified')
return resp()
@app.route("/setDefaults", methods=["POST"])
@auth.login_required
def setDefaults():
data = request.get_json()
username = data["username"]
log.debug(f"Storing default settings from user {username}")
try:
user = jf.getUsers(username=username, public=False)
except Jellyfin.UserNotFoundError:
log.error(f"Storing defaults failed: Couldn't find user {username}")
return resp(False)
uid = user["Id"]
policy = user["Policy"]
data_store.user_template = policy
if data["homescreen"]:
configuration = user["Configuration"]
try:
displayprefs = jf.getDisplayPreferences(uid)
data_store.user_configuration = configuration
data_store.user_displayprefs = displayprefs
except:
log.error("Storing defaults failed: " + "couldn't store homescreen layout")
return resp(False)
return resp()
import jellyfin_accounts.setup