DataOps-platform/deployment/app/commands/report_datasource_credentials.py

"""Read-only audit of legacy Neo4j DataSource credential state."""

import argparse
import json


SUPPORTED_TYPES = {"postgresql", "postgres", "mysql"}


def inspect_legacy_datasources(session):
    records = session.run(
        """
        MATCH (n:DataSource)
        RETURN elementId(n) AS legacy_id, properties(n) AS properties
        ORDER BY legacy_id
        """
    )
    report = []
    for record in records:
        properties = dict(record["properties"])
        issues = []
        uid = properties.get("uid")
        has_plaintext = any(
            properties.get(key)
            for key in (
                "username",
                "password",
                "conn_str",
                "connection_string",
                "connection_url",
            )
        )
        if not uid:
            issues.append("missing_uid")
        if has_plaintext:
            issues.append("plaintext_credentials")
        if (
            not has_plaintext
            and (
                not properties.get("credential_ref")
                or not properties.get("credential_version")
            )
        ):
            issues.append("missing_credential_reference")
        if str(properties.get("type") or "").lower() not in SUPPORTED_TYPES:
            issues.append("unsupported_database_type")
        if any(
            not properties.get(key)
            for key in ("host", "port", "database")
        ):
            issues.append("incomplete_connection_definition")
        if issues:
            report.append(
                {
                    "data_source_uid": str(uid) if uid else None,
                    "issues": issues,
                }
            )
    return report


def main():
    parser = argparse.ArgumentParser(
        description="Report legacy DataSource credential issues"
    )
    parser.parse_args()

    from app.services.neo4j_driver import neo4j_driver

    with neo4j_driver.get_session() as session:
        for item in inspect_legacy_datasources(session):
            print(json.dumps(item, ensure_ascii=False))


if __name__ == "__main__":
    main()