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