Build and return a mechanism specific credentials tuple.
(
mech: str,
source: Optional[str],
user: Optional[str],
passwd: Optional[str],
extra: Mapping[str, Any],
database: Optional[str],
)
| 98 | |
| 99 | |
| 100 | def _build_credentials_tuple( |
| 101 | mech: str, |
| 102 | source: Optional[str], |
| 103 | user: Optional[str], |
| 104 | passwd: Optional[str], |
| 105 | extra: Mapping[str, Any], |
| 106 | database: Optional[str], |
| 107 | ) -> MongoCredential: |
| 108 | """Build and return a mechanism specific credentials tuple.""" |
| 109 | if mech not in ("MONGODB-X509", "MONGODB-AWS", "MONGODB-OIDC") and user is None: |
| 110 | raise ConfigurationError(f"{mech} requires a username") |
| 111 | if mech == "GSSAPI": |
| 112 | if source is not None and source != "$external": |
| 113 | raise ValueError("authentication source must be $external or None for GSSAPI") |
| 114 | properties = extra.get("authmechanismproperties", {}) |
| 115 | service_name = properties.get("SERVICE_NAME", "mongodb") |
| 116 | service_host = properties.get("SERVICE_HOST", None) |
| 117 | canonicalize = properties.get("CANONICALIZE_HOST_NAME", "false") |
| 118 | canonicalize = _validate_canonicalize_host_name(canonicalize) |
| 119 | service_realm = properties.get("SERVICE_REALM") |
| 120 | props = GSSAPIProperties( |
| 121 | service_name=service_name, |
| 122 | canonicalize_host_name=canonicalize, |
| 123 | service_realm=service_realm, |
| 124 | service_host=service_host, |
| 125 | ) |
| 126 | # Source is always $external. |
| 127 | return MongoCredential(mech, "$external", user, passwd, props, None) |
| 128 | elif mech == "MONGODB-X509": |
| 129 | if passwd is not None: |
| 130 | raise ConfigurationError("Passwords are not supported by MONGODB-X509") |
| 131 | if source is not None and source != "$external": |
| 132 | raise ValueError("authentication source must be $external or None for MONGODB-X509") |
| 133 | # Source is always $external, user can be None. |
| 134 | return MongoCredential(mech, "$external", user, None, None, None) |
| 135 | elif mech == "MONGODB-AWS": |
| 136 | if user is not None and passwd is None: |
| 137 | raise ConfigurationError("username without a password is not supported by MONGODB-AWS") |
| 138 | if source is not None and source != "$external": |
| 139 | raise ConfigurationError( |
| 140 | "authentication source must be $external or None for MONGODB-AWS" |
| 141 | ) |
| 142 | |
| 143 | properties = extra.get("authmechanismproperties", {}) |
| 144 | aws_session_token = properties.get("AWS_SESSION_TOKEN") |
| 145 | aws_props = _AWSProperties(aws_session_token=aws_session_token) |
| 146 | # user can be None for temporary link-local EC2 credentials. |
| 147 | return MongoCredential(mech, "$external", user, passwd, aws_props, None) |
| 148 | elif mech == "MONGODB-OIDC": |
| 149 | properties = extra.get("authmechanismproperties", {}) |
| 150 | callback = properties.get("OIDC_CALLBACK") |
| 151 | human_callback = properties.get("OIDC_HUMAN_CALLBACK") |
| 152 | environ = properties.get("ENVIRONMENT") |
| 153 | token_resource = properties.get("TOKEN_RESOURCE", "") |
| 154 | default_allowed = [ |
| 155 | "*.mongodb.net", |
| 156 | "*.mongodb-dev.net", |
| 157 | "*.mongodb-qa.net", |