|
15 | 15 | from .gitlab_ci import GitLabCIDetector |
16 | 16 |
|
17 | 17 | if TYPE_CHECKING: |
| 18 | + from collections.abc import Mapping |
| 19 | + |
18 | 20 | from ... import CredentialContext |
19 | 21 |
|
20 | 22 | logger = logging.getLogger(__name__) |
|
30 | 32 | ] |
31 | 33 |
|
32 | 34 |
|
| 35 | +def registered_detectors() -> list[type[EnvironmentDetector]]: |
| 36 | + """Return the registered OIDC detectors in their default priority order.""" |
| 37 | + return list(_DETECTORS) |
| 38 | + |
| 39 | + |
| 40 | +def disable_env_var(identifier: str) -> str: |
| 41 | + """The environment variable that disables the detector with this id.""" |
| 42 | + return f"CLOUDSMITH_OIDC_{identifier.upper()}_DISABLED" |
| 43 | + |
| 44 | + |
| 45 | +def _is_disabled_value(value: str | None) -> bool: |
| 46 | + """Only the literal string ``true`` (case-insensitive) disables a detector.""" |
| 47 | + return (value or "").strip().lower() == "true" |
| 48 | + |
| 49 | + |
| 50 | +def disabled_detectors_from_env(environ: Mapping[str, str]) -> frozenset[str]: |
| 51 | + """Detector ids disabled via their CLOUDSMITH_OIDC_<ID>_DISABLED variable.""" |
| 52 | + return frozenset( |
| 53 | + detector_cls.id |
| 54 | + for detector_cls in _DETECTORS |
| 55 | + if _is_disabled_value(environ.get(disable_env_var(detector_cls.id))) |
| 56 | + ) |
| 57 | + |
| 58 | + |
| 59 | +def _ordered_detectors(order: str | None) -> list[type[EnvironmentDetector]]: |
| 60 | + """Detectors in evaluation order, honouring an explicit order string. |
| 61 | +
|
| 62 | + When ``order`` is unset/empty the default registration order is used. |
| 63 | + Otherwise only the listed ids are considered, in the listed order; |
| 64 | + unknown ids are logged and skipped. |
| 65 | + """ |
| 66 | + raw_order = (order or "").strip() |
| 67 | + if not raw_order: |
| 68 | + return list(_DETECTORS) |
| 69 | + |
| 70 | + detectors_by_id = {detector_cls.id: detector_cls for detector_cls in _DETECTORS} |
| 71 | + ordered: list[type[EnvironmentDetector]] = [] |
| 72 | + for token in raw_order.split(","): |
| 73 | + identifier = token.strip().lower() |
| 74 | + if not identifier: |
| 75 | + continue |
| 76 | + detector_cls = detectors_by_id.get(identifier) |
| 77 | + if detector_cls is None: |
| 78 | + logger.debug("Ignoring unknown OIDC detector id: %s", identifier) |
| 79 | + continue |
| 80 | + ordered.append(detector_cls) |
| 81 | + return ordered |
| 82 | + |
| 83 | + |
| 84 | +def _enabled_detectors( |
| 85 | + order: str | None, disabled: frozenset[str] |
| 86 | +) -> list[type[EnvironmentDetector]]: |
| 87 | + """Ordered detectors with disabled ones removed (disable always wins).""" |
| 88 | + return [ |
| 89 | + detector_cls |
| 90 | + for detector_cls in _ordered_detectors(order) |
| 91 | + if detector_cls.id not in disabled |
| 92 | + ] |
| 93 | + |
| 94 | + |
33 | 95 | def detect_environment( |
34 | 96 | context: CredentialContext, |
35 | 97 | ) -> EnvironmentDetector | None: |
36 | 98 | """Try each detector in order, returning the first that matches.""" |
37 | | - for detector_cls in _DETECTORS: |
| 99 | + enabled = _enabled_detectors( |
| 100 | + context.oidc_detector_order, context.oidc_disabled_detectors |
| 101 | + ) |
| 102 | + for detector_cls in enabled: |
38 | 103 | detector = detector_cls(context=context) |
39 | 104 | try: |
40 | 105 | if detector.detect(): |
|
0 commit comments