Validate authMechanismProperties.
(option: str, value: Any)
| 427 | |
| 428 | |
| 429 | def validate_auth_mechanism_properties(option: str, value: Any) -> dict[str, Union[bool, str]]: |
| 430 | """Validate authMechanismProperties.""" |
| 431 | props: dict[str, Any] = {} |
| 432 | if not isinstance(value, str): |
| 433 | if not isinstance(value, dict): |
| 434 | raise ValueError( |
| 435 | f"Auth mechanism properties must be given as a string or a dictionary, not {type(value)}" |
| 436 | ) |
| 437 | for key, value in value.items(): # noqa: B020 |
| 438 | if isinstance(value, str): |
| 439 | props[key] = value |
| 440 | elif isinstance(value, bool): |
| 441 | props[key] = str(value).lower() |
| 442 | elif key in ["ALLOWED_HOSTS"] and isinstance(value, list): |
| 443 | props[key] = value |
| 444 | elif key in ["OIDC_CALLBACK", "OIDC_HUMAN_CALLBACK"]: |
| 445 | from pymongo.auth_oidc_shared import OIDCCallback |
| 446 | |
| 447 | if not isinstance(value, OIDCCallback): |
| 448 | raise ValueError(f"callback must be an OIDCCallback object, not {type(value)}") |
| 449 | props[key] = value |
| 450 | else: |
| 451 | raise ValueError(f"Invalid type for auth mechanism property {key}, {type(value)}") |
| 452 | return props |
| 453 | |
| 454 | value = validate_string(option, value) |
| 455 | value = unquote_plus(value) |
| 456 | for opt in value.split(","): |
| 457 | key, _, val = opt.partition(":") |
| 458 | if not val: |
| 459 | raise ValueError("Malformed auth mechanism properties") |
| 460 | if key not in _MECHANISM_PROPS: |
| 461 | # Try not to leak the token. |
| 462 | if "AWS_SESSION_TOKEN" in key: |
| 463 | raise ValueError( |
| 464 | "auth mechanism properties must be " |
| 465 | "key:value pairs like AWS_SESSION_TOKEN:<token>" |
| 466 | ) |
| 467 | |
| 468 | raise ValueError( |
| 469 | f"{key} is not a supported auth " |
| 470 | "mechanism property. Must be one of " |
| 471 | f"{tuple(_MECHANISM_PROPS)}" |
| 472 | ) |
| 473 | |
| 474 | if key == "CANONICALIZE_HOST_NAME": |
| 475 | from pymongo.auth_shared import _validate_canonicalize_host_name |
| 476 | |
| 477 | props[key] = _validate_canonicalize_host_name(val) |
| 478 | else: |
| 479 | props[key] = val |
| 480 | |
| 481 | return props |
| 482 | |
| 483 | |
| 484 | def validate_document_class( |
nothing calls this directly
no test coverage detected