74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""Credential-free identity for the *connected* public-runtime PostgreSQL target.
|
|
|
|
The connection itself is the authority. A configured DSN is deliberately not
|
|
hashed because proxies, aliases, defaults, and credential changes can make a
|
|
URL identity disagree with the server that actually owns the inventory.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
|
|
DATABASE_IDENTITY_QUERY = """
|
|
SELECT
|
|
current_database()::text AS database_name,
|
|
current_user::text AS database_role,
|
|
COALESCE(inet_server_addr()::text, 'local') AS server_address,
|
|
inet_server_port() AS server_port
|
|
"""
|
|
|
|
|
|
def database_target_sha256(
|
|
*,
|
|
database_name: str,
|
|
database_role: str,
|
|
server_address: str,
|
|
server_port: int,
|
|
) -> str:
|
|
"""Return a lowercase SHA256 over a secret-free connected DB identity."""
|
|
|
|
if (
|
|
not isinstance(database_name, str)
|
|
or not database_name.strip()
|
|
or not isinstance(database_role, str)
|
|
or not database_role.strip()
|
|
or not isinstance(server_address, str)
|
|
or not server_address.strip()
|
|
):
|
|
raise ValueError("database target identity is incomplete")
|
|
if (
|
|
not isinstance(server_port, int)
|
|
or isinstance(server_port, bool)
|
|
or server_port < 1
|
|
or server_port > 65535
|
|
):
|
|
raise ValueError("database target port is invalid")
|
|
canonical = json.dumps(
|
|
{
|
|
"database": database_name,
|
|
"database_role": database_role,
|
|
"server_address": server_address,
|
|
"server_port": server_port,
|
|
},
|
|
ensure_ascii=True,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8", errors="strict")
|
|
return hashlib.sha256(canonical).hexdigest()
|
|
|
|
|
|
async def connected_database_target_sha256(connection: Any) -> str:
|
|
"""Read and hash the target identity through an already-open connection."""
|
|
|
|
target = await connection.fetchrow(DATABASE_IDENTITY_QUERY)
|
|
if target is None:
|
|
raise ValueError("database target identity is unavailable")
|
|
return database_target_sha256(
|
|
database_name=str(target["database_name"]),
|
|
database_role=str(target["database_role"]),
|
|
server_address=str(target["server_address"]),
|
|
server_port=int(target["server_port"]),
|
|
)
|