selfprivacy-rest-api/selfprivacy_api/utils/__init__.py

133 lines
3.6 KiB
Python
Raw Normal View History

2021-11-11 18:31:28 +00:00
#!/usr/bin/env python3
2021-11-16 16:14:01 +00:00
"""Various utility functions"""
2022-06-24 17:08:58 +00:00
import datetime
2022-01-14 05:38:53 +00:00
from enum import Enum
2021-11-22 16:50:50 +00:00
import json
import portalocker
2021-11-16 16:14:01 +00:00
2021-11-11 18:31:28 +00:00
USERDATA_FILE = "/etc/nixos/userdata/userdata.json"
2022-01-14 05:38:53 +00:00
TOKENS_FILE = "/etc/nixos/userdata/tokens.json"
DOMAIN_FILE = "/var/domain"
2021-11-30 21:53:39 +00:00
2022-01-14 05:38:53 +00:00
class UserDataFiles(Enum):
"""Enum for userdata files"""
USERDATA = 0
TOKENS = 1
2021-11-11 18:31:28 +00:00
def get_domain():
2021-11-16 16:14:01 +00:00
"""Get domain from /var/domain without trailing new line"""
with open(DOMAIN_FILE, "r", encoding="utf-8") as domain_file:
2021-11-16 16:14:01 +00:00
domain = domain_file.readline().rstrip()
2021-11-11 18:31:28 +00:00
return domain
2021-11-22 16:50:50 +00:00
class WriteUserData(object):
"""Write userdata.json with lock"""
2022-01-14 05:38:53 +00:00
def __init__(self, file_type=UserDataFiles.USERDATA):
if file_type == UserDataFiles.USERDATA:
self.userdata_file = open(USERDATA_FILE, "r+", encoding="utf-8")
elif file_type == UserDataFiles.TOKENS:
self.userdata_file = open(TOKENS_FILE, "r+", encoding="utf-8")
else:
raise ValueError("Unknown file type")
2021-11-22 16:50:50 +00:00
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"""
2022-01-14 05:38:53 +00:00
def __init__(self, file_type=UserDataFiles.USERDATA):
if file_type == UserDataFiles.USERDATA:
self.userdata_file = open(USERDATA_FILE, "r", encoding="utf-8")
elif file_type == UserDataFiles.TOKENS:
self.userdata_file = open(TOKENS_FILE, "r", encoding="utf-8")
else:
raise ValueError("Unknown file type")
2021-11-22 16:50:50 +00:00
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()
2021-11-23 18:32:51 +00:00
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
2022-06-24 17:08:58 +00:00
2022-06-24 18:14:20 +00:00
2022-06-24 17:08:58 +00:00
def parse_date(date_str: str) -> datetime.datetime:
"""Parse date string which can be in
%Y-%m-%dT%H:%M:%S.%fZ or %Y-%m-%d %H:%M:%S.%f format"""
2022-06-24 18:14:20 +00:00
return (
datetime.datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S.%fZ")
if date_str.endswith("Z")
else datetime.datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S.%f")
)