File size: 17,905 Bytes
1a7713e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 | """PostgreSQL access methods used by authentication and model tools.
Trusted application code should prefer narrow, parameterized methods such as
``get_customer_account_summary``. ``execute_query`` remains an intentionally raw
lab primitive and is not exposed to the Ollama model.
"""
import os
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any
from uuid import UUID
import psycopg
from dotenv import load_dotenv
class database_layer:
"""Own database configuration and the repository's PostgreSQL operations."""
def __init__(self):
"""Load local connection settings from ``configuration/.env``."""
self.database_env_file = Path(__file__).resolve().parents[1] / "configuration" / ".env"
# Explicitly load the repository-local file instead of relying on cwd.
load_dotenv(self.database_env_file)
# Defaults make the lab easy to start; deployments should override them.
self._connection_settings = {
"host": os.getenv("POSTGRES_HOST", "127.0.0.1"),
"port": os.getenv("POSTGRES_PORT", "5432"),
"dbname": os.getenv("POSTGRES_DB", "world_order_bank_system_db"),
"user": os.getenv("POSTGRES_USER", "username"),
"password": os.getenv("POSTGRES_PASSWORD", "1234")
}
# dotenv values are strings; currently even "False" is truthy here. The two
# execution paths are equivalent until firewall parsing and policy are implemented.
self._firewall_controls = os.getenv("firewall_controls")
print("Be sure that you configured your enviorment files correctly.")
def verify_customer_credentials(self, email: str, password: str) -> bool:
"""Check a customer's password without exposing its stored hash."""
# Parameters are bound separately so credentials never become SQL text.
query = """
SELECT EXISTS (
SELECT 1
FROM customers
WHERE email = %s
AND password_hash = crypt(%s, password_hash)
)
"""
with psycopg.connect(**self._connection_settings) as connection:
with connection.cursor() as cursor:
cursor.execute(query, (email, password))
result = cursor.fetchone()
return bool(result and result[0])
def get_customer_account_summary(self, email: str) -> dict[str, Any] | None:
"""Return permitted summary fields for exactly one authenticated email.
The caller—not the model—supplies ``email``. Returning ``None`` means that
no matching customer exists.
"""
if not isinstance(email, str) or not email.strip():
raise ValueError("The customer email cannot be empty.")
query = """
SELECT first_name, last_name, email, total_amount, currency_code
FROM customers
WHERE email = %s
"""
with psycopg.connect(**self._connection_settings) as connection:
with connection.cursor() as cursor:
cursor.execute(query, (email.strip(),))
result = cursor.fetchone()
if result is None:
return None
return {
"first_name": result[0],
"last_name": result[1],
"email": result[2],
"total_amount": result[3],
"currency_code": result[4],
}
# Convert model- or application-supplied amounts into exact PostgreSQL money
# values without rounding, accepting booleans, or carrying float artifacts.
@staticmethod
def _normalise_transfer_amount(amount: Decimal | str | int | float) -> Decimal:
"""Return a positive two-decimal amount suitable for ``NUMERIC(19, 2)``."""
# bool is an int subclass in Python, but True must never become a monetary 1.
if isinstance(amount, bool):
raise ValueError("The transfer amount must be a number.")
try:
# Convert through text so JSON numbers do not carry binary-float noise into SQL.
decimal_amount = Decimal(str(amount))
except (InvalidOperation, ValueError):
raise ValueError("The transfer amount must be a valid decimal number.") from None
if not decimal_amount.is_finite() or decimal_amount <= 0:
raise ValueError("The transfer amount must be greater than zero.")
try:
# Quantize for validation only; silently rounding money would change user intent.
normalised_amount = decimal_amount.quantize(Decimal("0.01"))
except InvalidOperation:
raise ValueError("The transfer amount is outside the supported range.") from None
if normalised_amount != decimal_amount:
raise ValueError("The transfer amount cannot have more than two decimal places.")
if normalised_amount > Decimal("99999999999999999.99"):
raise ValueError("The transfer amount is outside the supported range.")
return normalised_amount
# Validate a proposed transfer and create its pending ledger record. The sender
# is trusted application state, and this phase never changes customer balances.
def prepare_transfer(
self,
authenticated_sender_email: str,
recipient_email: str,
amount: Decimal | str | int | float,
) -> dict[str, Any]:
"""Create a pending transfer without changing either account balance."""
if not isinstance(authenticated_sender_email, str) or not authenticated_sender_email.strip():
raise ValueError("The authenticated sender email cannot be empty.")
if not isinstance(recipient_email, str) or not recipient_email.strip():
raise ValueError("The recipient email cannot be empty.")
sender_email = authenticated_sender_email.strip()
recipient_email = recipient_email.strip()
# The sender originates in authenticated application state, never in tool arguments.
if sender_email == recipient_email:
raise ValueError("The sender and recipient must be different customers.")
transfer_amount = self._normalise_transfer_amount(amount)
customer_query = """
SELECT customer_id, first_name, last_name, email, total_amount, currency_code
FROM customers
WHERE email IN (%s, %s)
"""
insert_query = """
INSERT INTO transfers (
sender_customer_id,
recipient_customer_id,
amount,
currency_code
)
VALUES (%s, %s, %s, %s)
RETURNING confirmation_id, status, created_at, expires_at
"""
with psycopg.connect(**self._connection_settings) as connection:
with connection.cursor() as cursor:
# Preparation reads both accounts but deliberately does not lock or update them.
# Confirmation repeats every mutable check under row locks.
cursor.execute(customer_query, (sender_email, recipient_email))
customers = {row[3]: row for row in cursor.fetchall()}
sender = customers.get(sender_email)
if sender is None:
raise ValueError("The authenticated sender account was not found.")
recipient = customers.get(recipient_email)
if recipient is None:
raise ValueError("The recipient account was not found.")
sender_currency = str(sender[5]).strip()
recipient_currency = str(recipient[5]).strip()
# Moving a numeric amount between currencies would invent an exchange rate.
if sender_currency != recipient_currency:
raise ValueError("Transfers between different currencies are not supported.")
if sender[4] < transfer_amount:
raise ValueError("The account has insufficient funds for this transfer.")
cursor.execute(
insert_query,
(sender[0], recipient[0], transfer_amount, sender_currency),
)
# This insert is the only write during preparation; balances are untouched.
transfer = cursor.fetchone()
if transfer is None:
raise RuntimeError("The pending transfer could not be created.")
return {
"confirmation_id": str(transfer[0]),
"status": transfer[1],
"sender_email": sender_email,
"recipient_email": recipient_email,
"recipient_name": f"{recipient[1]} {recipient[2]}",
"amount": transfer_amount,
"currency_code": sender_currency,
"created_at": transfer[2],
"expires_at": transfer[3],
}
# Complete one sender-owned pending transfer inside a single transaction. Row
# locks and repeated validation prevent replay and stale-balance updates.
def confirm_transfer(
self,
authenticated_sender_email: str,
confirmation_id: str,
) -> dict[str, Any]:
"""Atomically complete a pending transfer owned by the authenticated sender."""
if not isinstance(authenticated_sender_email, str) or not authenticated_sender_email.strip():
raise ValueError("The authenticated sender email cannot be empty.")
if not isinstance(confirmation_id, str) or not confirmation_id.strip():
raise ValueError("The confirmation ID cannot be empty.")
sender_email = authenticated_sender_email.strip()
try:
# Parse before querying so malformed identifiers never reach PostgreSQL.
parsed_confirmation_id = UUID(confirmation_id.strip())
except ValueError:
raise ValueError("The confirmation ID is not valid.") from None
transfer_query = """
SELECT
t.transfer_id,
t.status,
t.amount,
t.currency_code,
t.sender_customer_id,
t.recipient_customer_id,
sender.email,
recipient.email,
t.created_at,
t.expires_at,
t.confirmed_at
FROM transfers AS t
JOIN customers AS sender
ON sender.customer_id = t.sender_customer_id
JOIN customers AS recipient
ON recipient.customer_id = t.recipient_customer_id
WHERE t.confirmation_id = %s
AND sender.email = %s
FOR UPDATE OF t
"""
lock_customers_query = """
SELECT customer_id, total_amount, currency_code
FROM customers
WHERE customer_id IN (%s, %s)
ORDER BY customer_id
FOR UPDATE
"""
with psycopg.connect(**self._connection_settings) as connection:
with connection.cursor() as cursor:
# Locking the ledger row serializes concurrent confirmation and replay attempts.
# Scoping by sender email prevents one customer confirming another's transfer.
cursor.execute(transfer_query, (parsed_confirmation_id, sender_email))
transfer = cursor.fetchone()
if transfer is None:
raise ValueError(
"The transfer was not found for the authenticated customer."
)
transfer_id = transfer[0]
status = transfer[1]
transfer_amount = transfer[2]
transfer_currency = str(transfer[3]).strip()
sender_customer_id = transfer[4]
recipient_customer_id = transfer[5]
recipient_email = transfer[7]
if status == "completed":
# Replays report the previous outcome without moving money again.
return {
"confirmation_id": str(parsed_confirmation_id),
"status": "already_completed",
"sender_email": sender_email,
"recipient_email": recipient_email,
"amount": transfer_amount,
"currency_code": transfer_currency,
"confirmed_at": transfer[10],
}
if status == "expired":
return {
"confirmation_id": str(parsed_confirmation_id),
"status": "expired",
}
cursor.execute(
"""
UPDATE transfers
SET status = 'expired'
WHERE transfer_id = %s
AND status = 'pending'
AND expires_at <= CURRENT_TIMESTAMP
RETURNING status
""",
(transfer_id,),
)
if cursor.fetchone() is not None:
# Persist the terminal expired state while leaving balances unchanged.
return {
"confirmation_id": str(parsed_confirmation_id),
"status": "expired",
}
cursor.execute(
lock_customers_query,
(sender_customer_id, recipient_customer_id),
)
# Stable customer-ID ordering prevents opposite-direction transfers from
# taking the same pair of account locks in conflicting orders.
customer_rows = cursor.fetchall()
customers = {row[0]: row for row in customer_rows}
sender = customers.get(sender_customer_id)
recipient = customers.get(recipient_customer_id)
if sender is None or recipient is None:
raise RuntimeError("A transfer account no longer exists.")
sender_currency = str(sender[2]).strip()
recipient_currency = str(recipient[2]).strip()
# Balance and currency may have changed since preparation, so check again.
if (
sender_currency != transfer_currency
or recipient_currency != transfer_currency
):
raise ValueError("The account currencies no longer match the transfer.")
if sender[1] < transfer_amount:
raise ValueError("The account has insufficient funds for this transfer.")
cursor.execute(
"""
UPDATE customers
SET total_amount = total_amount - %s,
updated_at = CURRENT_TIMESTAMP
WHERE customer_id = %s
""",
(transfer_amount, sender_customer_id),
)
cursor.execute(
"""
UPDATE customers
SET total_amount = total_amount + %s,
updated_at = CURRENT_TIMESTAMP
WHERE customer_id = %s
""",
(transfer_amount, recipient_customer_id),
)
cursor.execute(
"""
UPDATE transfers
SET status = 'completed',
confirmed_at = CURRENT_TIMESTAMP
WHERE transfer_id = %s
AND status = 'pending'
RETURNING confirmed_at
""",
(transfer_id,),
)
completion = cursor.fetchone()
if completion is None:
raise RuntimeError("The transfer could not be confirmed.")
# The connection context commits all three updates together. Any exception
# rolls back both balance changes and the ledger status change.
return {
"confirmation_id": str(parsed_confirmation_id),
"status": "completed",
"sender_email": sender_email,
"recipient_email": recipient_email,
"amount": transfer_amount,
"currency_code": transfer_currency,
"confirmed_at": completion[0],
"remaining_balance": sender[1] - transfer_amount,
}
def execute_query(self, query:str) -> str:
"""Execute raw SQL through the selected lab path.
This intentionally dangerous primitive supports future firewall experiments.
It must not be exposed directly as an LLM tool.
"""
if(self._firewall_controls):
return self._secure_query_execute(query)
return self._non_secure_query_execute(query)
# TODO: enforce an actual SQL policy. This currently behaves like the unsafe path.
def _secure_query_execute(self, query:str ) -> str:
"""Execute SQL through the placeholder firewall-controlled path."""
with psycopg.connect(
**self._connection_settings
) as connection:
with connection.cursor() as cursor:
cursor.execute(query)
answer = cursor.fetchall()
return answer
def _non_secure_query_execute(self, query:str) -> str:
"""Execute unrestricted SQL for controlled pentesting experiments."""
with psycopg.connect(
**self._connection_settings
) as connection:
with connection.cursor() as cursor:
cursor.execute(query)
answer = cursor.fetchall()
return answer
|