|
3 | 3 | from __future__ import annotations |
4 | 4 |
|
5 | 5 | import logging |
| 6 | +import os |
6 | 7 | from typing import TYPE_CHECKING |
7 | 8 |
|
8 | 9 | from .aws import AWSDetector |
|
31 | 32 | GenericDetector, |
32 | 33 | ] |
33 | 34 |
|
| 35 | +DETECTOR_ORDER_ENV_VAR = "CLOUDSMITH_OIDC_DETECTOR_ORDER" |
| 36 | + |
| 37 | + |
| 38 | +def registered_detectors() -> list[type[EnvironmentDetector]]: |
| 39 | + """Return the registered OIDC detectors in their default priority order.""" |
| 40 | + return list(_DETECTORS) |
| 41 | + |
| 42 | + |
| 43 | +def _disable_env_var(slug: str) -> str: |
| 44 | + return f"CLOUDSMITH_OIDC_{slug.upper()}_DISABLED" |
| 45 | + |
| 46 | + |
| 47 | +def _is_detector_disabled(slug: str) -> bool: |
| 48 | + """Whether a detector is disabled via its CLOUDSMITH_OIDC_<SLUG>_DISABLED var. |
| 49 | +
|
| 50 | + Mirrors boto3's ``AWS_EC2_METADATA_DISABLED`` wording: only the literal |
| 51 | + string ``true`` (case-insensitive) disables; anything else is off. |
| 52 | + """ |
| 53 | + return os.environ.get(_disable_env_var(slug), "false").strip().lower() == "true" |
| 54 | + |
| 55 | + |
| 56 | +def _ordered_detectors() -> list[type[EnvironmentDetector]]: |
| 57 | + """Detectors in evaluation order, honouring CLOUDSMITH_OIDC_DETECTOR_ORDER. |
| 58 | +
|
| 59 | + When the order var is unset/empty the default registration order is used. |
| 60 | + Otherwise only the listed slugs are considered, in the listed order; |
| 61 | + unknown slugs are logged and skipped. |
| 62 | + """ |
| 63 | + raw_order = os.environ.get(DETECTOR_ORDER_ENV_VAR, "").strip() |
| 64 | + if not raw_order: |
| 65 | + return list(_DETECTORS) |
| 66 | + |
| 67 | + detectors_by_slug = {detector_cls.slug: detector_cls for detector_cls in _DETECTORS} |
| 68 | + ordered: list[type[EnvironmentDetector]] = [] |
| 69 | + for token in raw_order.split(","): |
| 70 | + slug = token.strip().lower() |
| 71 | + if not slug: |
| 72 | + continue |
| 73 | + detector_cls = detectors_by_slug.get(slug) |
| 74 | + if detector_cls is None: |
| 75 | + logger.debug( |
| 76 | + "Ignoring unknown OIDC detector slug in %s: %s", |
| 77 | + DETECTOR_ORDER_ENV_VAR, |
| 78 | + slug, |
| 79 | + ) |
| 80 | + continue |
| 81 | + ordered.append(detector_cls) |
| 82 | + return ordered |
| 83 | + |
| 84 | + |
| 85 | +def _enabled_detectors() -> list[type[EnvironmentDetector]]: |
| 86 | + """Ordered detectors with disabled ones removed (disable always wins).""" |
| 87 | + return [ |
| 88 | + detector_cls |
| 89 | + for detector_cls in _ordered_detectors() |
| 90 | + if not _is_detector_disabled(detector_cls.slug) |
| 91 | + ] |
| 92 | + |
34 | 93 |
|
35 | 94 | def detect_environment( |
36 | 95 | context: CredentialContext, |
37 | 96 | ) -> EnvironmentDetector | None: |
38 | 97 | """Try each detector in order, returning the first that matches.""" |
39 | | - for detector_cls in _DETECTORS: |
| 98 | + for detector_cls in _enabled_detectors(): |
40 | 99 | detector = detector_cls(context=context) |
41 | 100 | try: |
42 | 101 | if detector.detect(): |
|
0 commit comments