"""Secrets Manager access for sanitisation. PostgreSQL role passwords cannot be migrated from the source: on RDS the master user is rds_superuser, not a true superuser, and cannot read pg_authid/pg_shadow (see src.logic.postgresql). Instead the known dev/QA passwords are stored in a per-target-database Secrets Manager secret and applied during copy_users. The secret is named {prefix}{stable_db_name}/{suffix}, where the prefix is config.ROLE_PASSWORDS_SECRET_NAME_PREFIX and stable_db_name is the target db name with any intermediate -tmp-{timestamp} suffix stripped (sanitisation runs against the temporary restored cluster, but the secret is keyed on the stable name). The per-database name lets one deployment that sanitises several databases in the same account keep a separate secret for each. The prefix is optional: when unset (e.g. MySQL-only accounts) no secret is read and roles are left passwordless. Its SecretString is a JSON object mapping role name to password, e.g. {"app_login": "...", "app_owner": "..."} """ import json import re from lambdacommon.common_config import logger import config from src.connectors.boto_clients import secrets_manager_client as sm # During a refresh the target is restored to an intermediate cluster named # {db_name}-tmp-{YYYYMMDDHHMMSS} (see fargate/restore) and sanitised before # being renamed to {db_name}. The role-password secret is keyed on the stable # db_name, so strip any such suffix before building the secret id. _TMP_SUFFIX_RE = re.compile(r'-tmp-\d{14}$') def get_role_passwords(target_db_name): """Return a {role_name: password} map for the target database. Reads the secret named {prefix}{stable_db_name}/{suffix} (see the module docstring; any intermediate -tmp-{timestamp} suffix on target_db_name is stripped first) and parses its JSON SecretString. Returns an empty dict when no prefix is configured or the named secret does not exist, so accounts without one (MySQL-only or otherwise) keep the existing passwordless behaviour. Password values are never logged. """ prefix = config.ROLE_PASSWORDS_SECRET_NAME_PREFIX if not prefix: logger.info( 'No ROLE_PASSWORDS_SECRET_NAME_PREFIX configured; roles will be ' 'left passwordless.') return {} stable_db_name = _TMP_SUFFIX_RE.sub('', target_db_name) secret_id = ( f'{prefix}{stable_db_name}/{config.ROLE_PASSWORDS_SECRET_SUFFIX}') try: secret = sm.get_secret_value(SecretId=secret_id) except sm.exceptions.ResourceNotFoundException: logger.info( f'No role-password secret {secret_id}; roles will be left ' 'passwordless.') return {} passwords = json.loads(secret['SecretString']) logger.info( f'Loaded passwords for {len(passwords)} roles from {secret_id}') return passwords