mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-11-14 22:53:17 +00:00
13cef67204
Reviewed-on: https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api/pulls/7 Co-authored-by: Inex Code <inex.code@selfprivacy.org> Co-committed-by: Inex Code <inex.code@selfprivacy.org>
103 lines
2.5 KiB
Python
103 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Various utility functions"""
|
|
import json
|
|
import portalocker
|
|
|
|
|
|
USERDATA_FILE = "/etc/nixos/userdata/userdata.json"
|
|
DOMAIN_FILE = "/var/domain"
|
|
|
|
|
|
def get_domain():
|
|
"""Get domain from /var/domain without trailing new line"""
|
|
with open(DOMAIN_FILE, "r", encoding="utf-8") as domain_file:
|
|
domain = domain_file.readline().rstrip()
|
|
return domain
|
|
|
|
|
|
class WriteUserData(object):
|
|
"""Write userdata.json with lock"""
|
|
|
|
def __init__(self):
|
|
self.userdata_file = open(USERDATA_FILE, "r+", encoding="utf-8")
|
|
portalocker.lock(self.userdata_file, portalocker.LOCK_EX)
|
|
self.data = json.load(self.userdata_file)
|
|
|
|
def __enter__(self):
|
|
return self.data
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
if exc_type is None:
|
|
self.userdata_file.seek(0)
|
|
json.dump(self.data, self.userdata_file, indent=4)
|
|
self.userdata_file.truncate()
|
|
portalocker.unlock(self.userdata_file)
|
|
self.userdata_file.close()
|
|
|
|
|
|
class ReadUserData(object):
|
|
"""Read userdata.json with lock"""
|
|
|
|
def __init__(self):
|
|
self.userdata_file = open(USERDATA_FILE, "r", encoding="utf-8")
|
|
portalocker.lock(self.userdata_file, portalocker.LOCK_SH)
|
|
self.data = json.load(self.userdata_file)
|
|
|
|
def __enter__(self):
|
|
return self.data
|
|
|
|
def __exit__(self, *args):
|
|
portalocker.unlock(self.userdata_file)
|
|
self.userdata_file.close()
|
|
|
|
|
|
def validate_ssh_public_key(key):
|
|
"""Validate SSH public key. It may be ssh-ed25519 or ssh-rsa."""
|
|
if not key.startswith("ssh-ed25519"):
|
|
if not key.startswith("ssh-rsa"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def is_username_forbidden(username):
|
|
forbidden_prefixes = ["systemd", "nixbld"]
|
|
|
|
forbidden_usernames = [
|
|
"root",
|
|
"messagebus",
|
|
"postfix",
|
|
"polkituser",
|
|
"dovecot2",
|
|
"dovenull",
|
|
"nginx",
|
|
"postgres",
|
|
"prosody",
|
|
"opendkim",
|
|
"rspamd",
|
|
"sshd",
|
|
"selfprivacy-api",
|
|
"restic",
|
|
"redis",
|
|
"pleroma",
|
|
"ocserv",
|
|
"nextcloud",
|
|
"memcached",
|
|
"knot-resolver",
|
|
"gitea",
|
|
"bitwarden_rs",
|
|
"vaultwarden",
|
|
"acme",
|
|
"virtualMail",
|
|
"nobody",
|
|
]
|
|
|
|
for prefix in forbidden_prefixes:
|
|
if username.startswith(prefix):
|
|
return True
|
|
|
|
for forbidden_username in forbidden_usernames:
|
|
if username == forbidden_username:
|
|
return True
|
|
|
|
return False
|