328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""Database and encryption services for the legacy Flowers data format.
|
|
|
|
The public ``Flower`` API intentionally remains compatible with the original
|
|
application. Existing tables and Fernet ciphertext require no migration.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from contextlib import contextmanager
|
|
from typing import Callable, Iterator
|
|
|
|
import pymysql as mdb
|
|
from fernet import Fernet
|
|
|
|
from Config import (
|
|
DB_ADMIN_PASSWORD,
|
|
DB_ADMIN_USER,
|
|
DB_HOST,
|
|
DB_NAME,
|
|
DB_PORT,
|
|
MINFIELDLEN,
|
|
)
|
|
from Log import Log
|
|
|
|
|
|
_USERNAME = re.compile(r"^[A-Za-z0-9_]{1,31}$")
|
|
|
|
|
|
class InvalidCredentials(ValueError):
|
|
"""The supplied username/password cannot address a Flowers vault."""
|
|
|
|
|
|
class StorageError(RuntimeError):
|
|
"""A database operation failed."""
|
|
|
|
|
|
def legacy_key(password: str) -> str:
|
|
"""Return the exact DB password/Fernet key used by historical releases."""
|
|
if not isinstance(password, str) or not password:
|
|
raise InvalidCredentials("A password is required")
|
|
key = (password * 10)[:43] + "="
|
|
try:
|
|
Fernet(key.encode("ascii"))
|
|
except (UnicodeEncodeError, ValueError) as exc:
|
|
raise InvalidCredentials(
|
|
"Password must use Base64-compatible characters"
|
|
) from exc
|
|
return key
|
|
|
|
|
|
def _identifier(username: str) -> str:
|
|
if not isinstance(username, str) or not _USERNAME.fullmatch(username):
|
|
raise InvalidCredentials(
|
|
"Username must contain only letters, numbers, and underscores"
|
|
)
|
|
return f"{username}_"
|
|
|
|
|
|
class Flower:
|
|
"""Read and write one user's existing Flowers table."""
|
|
|
|
def __init__(
|
|
self,
|
|
user: str,
|
|
pwd: str,
|
|
connection_factory: Callable[..., object] | None = None,
|
|
) -> None:
|
|
self.db = DB_NAME
|
|
self.db_username = _identifier(user)
|
|
self.db_password = legacy_key(pwd)
|
|
self.customer_table = self.db_username
|
|
self._table = f"`{self.customer_table}`"
|
|
self.fernet = Fernet(self.db_password.encode("ascii"))
|
|
self._connect = connection_factory or mdb.connect
|
|
|
|
@contextmanager
|
|
def _connection(self) -> Iterator[object]:
|
|
connection = None
|
|
try:
|
|
connection = self._connect(
|
|
host=DB_HOST,
|
|
port=DB_PORT,
|
|
user=self.db_username,
|
|
passwd=self.db_password,
|
|
db=self.db,
|
|
charset="latin1",
|
|
)
|
|
yield connection
|
|
except mdb.Error as exc:
|
|
Log.debug(f"Database error: {exc}")
|
|
raise StorageError("The vault database is unavailable") from exc
|
|
finally:
|
|
if connection is not None:
|
|
connection.close()
|
|
|
|
def encode(self, value: str) -> str:
|
|
if not value:
|
|
return ""
|
|
return self.fernet.encrypt(value.encode("utf-8")).decode("ascii")
|
|
|
|
def decode(self, value: str) -> str:
|
|
if not value:
|
|
return ""
|
|
return self.fernet.decrypt(value.encode("ascii")).decode("utf-8")
|
|
|
|
def authenticate(self) -> bool:
|
|
try:
|
|
with self._connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {self._table}")
|
|
cursor.fetchone()
|
|
return True
|
|
except (StorageError, ValueError):
|
|
return False
|
|
|
|
def numberOfEntries(self) -> int:
|
|
"""Compatibility method used by older clients and scripts."""
|
|
try:
|
|
with self._connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {self._table}")
|
|
row = cursor.fetchone()
|
|
return int(row[0]) if row else 0
|
|
except StorageError:
|
|
return -1
|
|
|
|
def add(
|
|
self,
|
|
organization: str,
|
|
myname: str,
|
|
myid: str,
|
|
secret: str,
|
|
deleted: int = 0,
|
|
timestamp=None,
|
|
) -> dict:
|
|
item = self._item(organization, "FAILED", myid, myname, secret)
|
|
if not all(isinstance(v, str) for v in (organization, myname, myid, secret)):
|
|
return item
|
|
if not (
|
|
len(organization) > MINFIELDLEN
|
|
and len(myid) > MINFIELDLEN
|
|
and len(secret) > MINFIELDLEN
|
|
):
|
|
return item
|
|
|
|
columns = "organization, myID, myName, mySecret, deleted"
|
|
values = [
|
|
organization,
|
|
self.encode(myid),
|
|
self.encode(myname),
|
|
self.encode(secret),
|
|
int(bool(deleted)),
|
|
]
|
|
placeholders = "%s, %s, %s, %s, %s"
|
|
if timestamp is not None:
|
|
columns += ", dateCreated"
|
|
placeholders += ", %s"
|
|
values.append(timestamp)
|
|
|
|
sql = f"INSERT INTO {self._table} ({columns}) VALUES ({placeholders})"
|
|
self._execute_write(sql, values)
|
|
return self._item(organization, "STORED", myid, myname, secret)
|
|
|
|
def all(self, isdeleted: int = 0) -> list[dict]:
|
|
sql = f"""
|
|
SELECT latest.organization, latest.myID, latest.dateCreated
|
|
FROM {self._table} AS latest
|
|
INNER JOIN (
|
|
SELECT organization, MAX(dateCreated) AS lastCreated
|
|
FROM {self._table}
|
|
GROUP BY organization
|
|
) AS versions
|
|
ON versions.organization = latest.organization
|
|
AND versions.lastCreated = latest.dateCreated
|
|
WHERE latest.deleted = %s
|
|
ORDER BY latest.dateCreated DESC, latest.organization ASC
|
|
"""
|
|
with self._connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(sql, (int(bool(isdeleted)),))
|
|
rows = cursor.fetchall()
|
|
return [
|
|
{
|
|
"organization": str(row[0]),
|
|
"myID": self.decode(row[1]),
|
|
"dateCreated": str(row[2]),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def one(self, organization: str, datetimestamp) -> dict:
|
|
sql = f"""
|
|
SELECT mySecret, myID, myName
|
|
FROM {self._table}
|
|
WHERE organization = %s AND dateCreated = %s
|
|
ORDER BY dateCreated DESC
|
|
LIMIT 1
|
|
"""
|
|
with self._connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(sql, (organization, datetimestamp))
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return self._item(
|
|
organization,
|
|
str(datetimestamp),
|
|
self.decode(row[1]),
|
|
self.decode(row[2]),
|
|
self.decode(row[0]),
|
|
)
|
|
return self._item(organization, str(datetimestamp), "", "", "-- Not found --")
|
|
|
|
def deactivate(self, organization: str, datetimestamp) -> dict:
|
|
sql = f"UPDATE {self._table} SET deleted = 1 WHERE organization = %s AND dateCreated = %s"
|
|
changed = self._execute_write(sql, (organization, datetimestamp))
|
|
message = "- DELETED -" if changed else "-- Not found --"
|
|
return self._item(organization, str(datetimestamp), "", "", message)
|
|
|
|
def empty(self) -> dict:
|
|
return self._item("", "", "", "", "")
|
|
|
|
def update_pwd(self, new_pwd: str) -> str:
|
|
"""Re-encrypt every value and update the legacy per-user DB password."""
|
|
new_db_password = legacy_key(new_pwd)
|
|
new_fernet = Fernet(new_db_password.encode("ascii"))
|
|
|
|
try:
|
|
with self._connection() as connection:
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
f"SELECT myID, myName, mySecret, organization, dateCreated FROM {self._table}"
|
|
)
|
|
rows = cursor.fetchall()
|
|
update = f"""
|
|
UPDATE {self._table}
|
|
SET myID = %s, myName = %s, mySecret = %s
|
|
WHERE organization = %s AND dateCreated = %s
|
|
"""
|
|
for row in rows:
|
|
encrypted = [
|
|
new_fernet.encrypt(self.decode(value).encode("utf-8")).decode("ascii")
|
|
if value else ""
|
|
for value in row[:3]
|
|
]
|
|
cursor.execute(update, (*encrypted, row[3], row[4]))
|
|
# This form is supported by the MariaDB versions used by
|
|
# historical Flowers installations.
|
|
cursor.execute("SET PASSWORD = PASSWORD(%s)", (new_db_password,))
|
|
connection.commit()
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
except Exception as exc:
|
|
Log.debug(f"Password update failed: {exc}")
|
|
return "ERROR updating password"
|
|
return "SUCCESSFULLY updated, now login again"
|
|
|
|
def _execute_write(self, sql: str, parameters) -> int:
|
|
with self._connection() as connection:
|
|
try:
|
|
with connection.cursor() as cursor:
|
|
changed = cursor.execute(sql, parameters)
|
|
connection.commit()
|
|
return int(changed)
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
|
|
@staticmethod
|
|
def _item(organization, timestamp, myid, myname, secret) -> dict:
|
|
return {
|
|
"organization": organization,
|
|
"dateCreated": timestamp,
|
|
"myID": myid,
|
|
"myName": myname,
|
|
"mySecret": secret,
|
|
}
|
|
|
|
|
|
class SuperFlower(Flower):
|
|
"""Provision a legacy per-user table and database account."""
|
|
|
|
def __init__(self, customer: str, pwd: str, connection_factory=None) -> None:
|
|
super().__init__(customer, pwd, connection_factory)
|
|
self.customer_name = self.customer_table
|
|
self.customer_pwd = self.db_password
|
|
self.db_username = DB_ADMIN_USER
|
|
self.db_password = DB_ADMIN_PASSWORD
|
|
|
|
def createNewTable(self) -> int:
|
|
statements = [
|
|
f"""CREATE TABLE {self._table} (
|
|
organization varchar(80) NOT NULL,
|
|
myID varchar(255) NOT NULL,
|
|
myName varchar(255) NOT NULL,
|
|
mySecret varchar(255) NOT NULL,
|
|
dateCreated timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
deleted int(11) DEFAULT 0,
|
|
PRIMARY KEY (organization, dateCreated)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=latin1""",
|
|
f"CREATE USER '{self.customer_name}'@'localhost' IDENTIFIED BY %s",
|
|
f"GRANT SELECT, UPDATE, INSERT ON `{self.db}`.{self._table} TO '{self.customer_name}'@'localhost'",
|
|
]
|
|
try:
|
|
with self._connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(statements[0])
|
|
cursor.execute(statements[1], (self.customer_pwd,))
|
|
cursor.execute(statements[2])
|
|
connection.commit()
|
|
return 1
|
|
except StorageError as exc:
|
|
Log.info(f"Could not create vault: {exc}")
|
|
return 0
|
|
|
|
def flush_privs(self) -> int:
|
|
try:
|
|
self._execute_write("FLUSH PRIVILEGES", ())
|
|
return 1
|
|
except StorageError:
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("This module is a library.")
|