protocol_postgres.py

"""
PostgreSQL login testing via asyncpg. Postgres returns a specific
SQLSTATE (28P01, invalid_password, or 28000, invalid_authorization_specification)
for bad credentials, distinct from a connection-level failure.
"""

import time

import asyncpg


class PostgresResult:
    def __init__(self, classification, evidence, elapsed):
        self.classification = classification
        self.evidence = evidence
        self.elapsed = elapsed


async def try_login(host, port, username, password, database="postgres", timeout=8):
    start = time.monotonic()
    try:
        conn = await asyncpg.connect(host=host, port=port, user=username, password=password, database=database, timeout=timeout)
        elapsed = time.monotonic() - start
        await conn.close()
        return PostgresResult("possible_success", "PostgreSQL authentication succeeded, a connection was established", elapsed)
    except asyncpg.InvalidPasswordError as exc:
        return PostgresResult("fail", f"server returned invalid_password (28P01): {exc}", time.monotonic() - start)
    except asyncpg.InvalidAuthorizationSpecificationError as exc:
        return PostgresResult("fail", f"server returned invalid_authorization_specification (28000): {exc}", time.monotonic() - start)
    except (asyncpg.TooManyConnectionsError, asyncpg.ConnectionDoesNotExistError) as exc:
        return PostgresResult("rate_limit", f"connection limit or resource issue: {exc}", time.monotonic() - start)
    except Exception as exc:
        return PostgresResult("rate_limit", f"connection error: {exc}", time.monotonic() - start)