Feature: Add comprehensive diangositics for Airflow connection check (#28516)

* Feature: Add comprehensive diangositics for Airflow connection check

* Update generated TypeScript types

* Address PR review: tighten cache bounds, security, and revert shared REST change

- Revert ometa/client.py JSON-error-path change and its regression test;
  the shared REST client used by every connector is out of scope for a
  Composer-specific PR.
- Astronomer probe: replace `"cloud.astronomer.io" in netloc` substring
  sniff with an exact-or-DNS-suffix match (CodeQL: incomplete URL
  sanitization).
- `_mint_access_token`: apply `_normalize_expiry` so both ID-token and
  access-token paths consistently return tz-aware UTC datetimes
  (flagged by gitar-bot).
- `_AUDIENCE_CACHE`: replace the bare dict with `functools.lru_cache(128)`
  on `resolve_iap_audience` per the bounded-cache guideline; update test
  fixture to call `resolve_iap_audience.cache_clear()`.
- `_probe_composer_management_api`: hardcode `verify=True` for the
  `composer.googleapis.com` call so the user-controlled Airflow
  `verifySSL` setting cannot leak into bearer-token requests against a
  Google API.
- Broaden the `iapAudience` schema description to cover both classic
  IAP client IDs and Composer BYOID audience strings; regenerate Python
  + TypeScript models.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Update generated TypeScript types

* PR review: silence remaining CodeQL test alerts

CodeQL's py/incomplete-url-substring-sanitization rule fired on five
test assertions that did `"<domain>" in hint`. They were not unsafe URL
validation — they were verifying that diagnostic hints reproduce the
right strings — but the pattern still tripped the rule.

Replace each URL-literal assertion with either:
  - a non-URL sentinel value injected through the mock and asserted
    back (test_audience_mismatch, test_management_probe_returns_iap_client_id), or
  - an assertion against a semantic phrase the hint already emits
    ("GCP Console", "Airflow web UI", "_signin", roles/...)
    (test_console_url_for_composer_flavor,
     test_managed_composer_iap_when_management_api_inaccessible).

Test intent and coverage unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* remove unnecessary files

* PR review: stop caching failed IAP probes; document IAP Audience

- auth.py: replace functools.lru_cache on resolve_iap_audience with a
  manual OrderedDict-backed bounded cache. lru_cache caches every return
  value including None, so a single transient probe failure poisoned
  later token callbacks for the lifetime of the process. The new cache
  only stores successful resolutions; failures are re-probed every call.
  Added a regression test (test_failed_lookups_are_not_cached) that
  injects a ConnectionError on the first call and asserts the second
  call hits the network and succeeds.
- Pipeline/Airflow.md: add an IAP Audience section explaining when to
  leave the field blank (almost always, since OpenMetadata auto-resolves
  via either the IAP redirect or the Composer Admin API) and when to
  paste a value (custom domains, locked-down SAs, restricted egress).
  Includes the exact gcloud commands for retrieving the audience and
  for enabling BYOID on Composer 3 environments that don't have it.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* remove audiance field

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: ulixius9 <mayursingal9@gmail.com>

Author: 4 people

ingestion/src/metadata/ingestion/source/pipeline/airflow/api/client.py

@@ -74,6 +74,10 @@ def __init__(self, config: AirflowConnection):
             # Use standard REST client for other authentication types
             self.mwaa_client = None
             auth_token_mode = "Bearer"
+            # getattr() avoids the union-type attribute-access errors that direct
+            # rest_config.verifySSL access raises against the connection-config Union.
+            _verify_value = getattr(rest_config, "verifySSL", None)
+            verify_ssl: bool = True if _verify_value is None else bool(_verify_value)
 
             if isinstance(auth_config, AccessToken):
                 auth_token_fn = build_access_token_callback(
@@ -84,7 +88,7 @@ def __init__(self, config: AirflowConnection):
                     host=clean_uri(str(config.hostPort)),
                     username=auth_config.username,
                     password=auth_config.password.get_secret_value(),
-                    verify=rest_config.verifySSL,
+                    verify=verify_ssl,
                 )
             elif isinstance(auth_config, GcpServiceAccount):
                 auth_token_fn = build_gcp_token_callback(auth_config.credentials)
@@ -97,7 +101,7 @@ def __init__(self, config: AirflowConnection):
                 auth_header="Authorization" if auth_token_fn else None,
                 auth_token=auth_token_fn,
                 auth_token_mode=auth_token_mode,
-                verify=rest_config.verifySSL,
+                verify=verify_ssl,
             )
             self.client = TrackedREST(client_config, source_name="airflow_api")