100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import secrets
|
|
|
|
from cryptography.fernet import Fernet
|
|
from passlib.hash import sha512_crypt
|
|
|
|
|
|
# The installer requires a Linux crypt hash. The builtin backend also works
|
|
# on Python 3.13+, where the standard-library crypt module no longer exists.
|
|
_installer_password_hash = sha512_crypt.using(rounds=656000)
|
|
_installer_password_hash.set_backend("builtin")
|
|
|
|
|
|
def canonical(value):
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
|
|
|
|
|
|
def digest(value):
|
|
if not isinstance(value, bytes):
|
|
value = value.encode()
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def token():
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
class Security:
|
|
def __init__(self, settings):
|
|
path = settings.master_key_file
|
|
if not path.exists():
|
|
if not settings.testing:
|
|
raise RuntimeError("Master-Key fehlt. Zuerst 'provisioner init' ausführen.")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(Fernet.generate_key())
|
|
path.chmod(0o600)
|
|
self.fernet = Fernet(path.read_bytes().strip())
|
|
|
|
@staticmethod
|
|
def hash_password(password):
|
|
if len(password) < 12 or len(password) > 1024:
|
|
raise ValueError("Passwörter benötigen 12 bis 1024 Zeichen.")
|
|
salt = os.urandom(16)
|
|
result = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1)
|
|
return "scrypt$" + base64.b64encode(salt).decode() + "$" + base64.b64encode(result).decode()
|
|
|
|
@staticmethod
|
|
def hash_installer_password(password):
|
|
if not 12 <= len(password) <= 1024 or "\x00" in password:
|
|
raise ValueError("Root-Passwörter benötigen 12 bis 1024 Zeichen ohne Nullzeichen.")
|
|
return _installer_password_hash.hash(password)
|
|
|
|
@staticmethod
|
|
def verify_password(password, stored):
|
|
try:
|
|
algorithm, salt, expected = stored.split("$")
|
|
if algorithm != "scrypt" or len(password) > 1024:
|
|
return False
|
|
result = hashlib.scrypt(password.encode(), salt=base64.b64decode(salt), n=16384, r=8, p=1)
|
|
return hmac.compare_digest(result, base64.b64decode(expected))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
def encrypt(self, value):
|
|
return self.fernet.encrypt(value.encode()).decode()
|
|
|
|
def decrypt(self, value):
|
|
return self.fernet.decrypt(value.encode()).decode()
|
|
|
|
|
|
def redact(text, values=()):
|
|
text = str(text)
|
|
for value in sorted({str(v) for v in values if v}, key=len, reverse=True):
|
|
text = text.replace(value, "[REDACTED]")
|
|
text = re.sub(r"(?i)(bearer\s+)[^\s\"']+", r"\1[REDACTED]", text)
|
|
text = re.sub(r"(/(?:bootstrap/v1|installer/v1/report)/)[A-Za-z0-9_-]+", r"\1[REDACTED]", text)
|
|
text = re.sub(r"(?i)((?:password|secret|token|authorization)\s*[=:]\s*)[^\s,;]+", r"\1[REDACTED]", text)
|
|
return text
|
|
|
|
|
|
def atomic_artifact(directory: Path, source: str):
|
|
content = source.encode("utf-8")
|
|
checksum = digest(content)
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
target = directory / checksum
|
|
if not target.exists():
|
|
temporary = directory / (".tmp-" + token())
|
|
with temporary.open("xb") as stream:
|
|
stream.write(content)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, target)
|
|
return checksum
|