AAP-68113 [PENDING] Candle pin v2

metrics_utility/automation_controller_billing/collector.py

@@ -71,10 +71,14 @@ def _gather_config(self):
         if not super()._gather_config():
             return False
 
-        # Extend the config collection to contain billing specific info:
+        # Extend the config collection to contain billing specific info.
+        # Strip mTLS key material — it is only needed at ship time and must never be
+        # included in the payload archive that is uploaded to the remote ingress endpoint.
+        _SENSITIVE_BILLING_KEYS = {'candlepin_cert_pem', 'candlepin_key_pem'}
+        safe_billing_provider_params = {k: v for k, v in (self.billing_provider_params or {}).items() if k not in _SENSITIVE_BILLING_KEYS}
         config_collection = self.collections['config']
         data = json.loads(config_collection.data)
-        data['billing_provider_params'] = self.billing_provider_params
+        data['billing_provider_params'] = safe_billing_provider_params
         config_collection._save_gathering(data)
 
         return True

metrics_utility/automation_controller_billing/package/package_crc.py

@@ -9,6 +9,8 @@
 import metrics_utility.base as base
 
 from metrics_utility.exceptions import FailedToUploadPayload
+from metrics_utility.library.candlepin.client import CandlepinClient
+from metrics_utility.library.candlepin.lifecycle import is_cert_valid as _is_cert_valid
 from metrics_utility.logger import logger
 
 
@@ -18,6 +20,18 @@ class PackageCRC(base.Package):
 
     SHIPPING_AUTH_SERVICE_ACCOUNT = 'service-account'
 
+    def __init__(self, collector):
+        super().__init__(collector)
+        self._resolved_auth_mode = None
+        self._temp_cert_path = None
+        self._temp_key_path = None
+
+    def _candlepin_cert_pem(self):
+        return (self.collector.billing_provider_params or {}).get('candlepin_cert_pem')
+
+    def _candlepin_key_pem(self):
+        return (self.collector.billing_provider_params or {}).get('candlepin_key_pem')
+
     def _tarname_base(self):
         timestamp = self.collector.gather_until
         return f'{settings.SYSTEM_UUID}-{timestamp.strftime("%Y-%m-%d-%H%M%S%z")}'
@@ -40,19 +54,67 @@ def _get_rh_password(self):
     def _get_http_request_headers(self):
         return get_awx_http_client_headers()
 
+    def _get_client_certificates(self):
+        if self._temp_cert_path and self._temp_key_path:
+            return (self._temp_cert_path, self._temp_key_path)
+        return super()._get_client_certificates()
+
+    def ship(self):
+        if self.shipping_auth_mode() == self.SHIPPING_AUTH_SERVICE_ACCOUNT:
+            return super().ship()
+
+        # mTLS path: delegate temp-file lifecycle to CandlepinClient._temp_cert_files so
+        # the files exist for the duration of the POST then are unconditionally cleaned up.
+        try:
+            with CandlepinClient._temp_cert_files(self._candlepin_cert_pem(), self._candlepin_key_pem()) as (cert_path, key_path):
+                self._temp_cert_path = cert_path
+                self._temp_key_path = key_path
+                try:
+                    return super().ship()
+                except requests.exceptions.SSLError as e:
+                    # Before falling back, verify that service account credentials are present.
+                    # In an mTLS-only deployment they won't be, and a blind retry would silently
+                    # return False with no actionable error context for the operator.
+                    if not self._get_rh_user() or not self._get_rh_password():
+                        raise FailedToUploadPayload(
+                            f'mTLS upload failed and no service account credentials are configured to fall back to '
+                            f'(METRICS_UTILITY_SERVICE_ACCOUNT_ID / METRICS_UTILITY_SERVICE_ACCOUNT_SECRET are not set). '
+                            f'Original SSL error: {e}'
+                        ) from e
+                    logger.error(f'mTLS upload failed ({e}); retrying with service account auth')
+                    self._resolved_auth_mode = self.SHIPPING_AUTH_SERVICE_ACCOUNT
+                    self._temp_cert_path = None
+                    self._temp_key_path = None
+                    return super().ship()
+        finally:
+            self._temp_cert_path = None
+            self._temp_key_path = None
+
     def shipping_auth_mode(self):
-        # TODO make this as a configuration so we can use this for local testing,
-        # for now, uncomment when testin locally in docker
-        # return self.SHIPPING_AUTH_IDENTITY
+        if self._resolved_auth_mode is not None:
+            return self._resolved_auth_mode
 
-        return self.SHIPPING_AUTH_SERVICE_ACCOUNT
+        cert_pem = self._candlepin_cert_pem()
+        key_pem = self._candlepin_key_pem()
+        if cert_pem and key_pem and _is_cert_valid(cert_pem):
+            self._resolved_auth_mode = self.SHIPPING_AUTH_CERTIFICATES
+        else:
+            self._resolved_auth_mode = self.SHIPPING_AUTH_SERVICE_ACCOUNT
+
+        return self._resolved_auth_mode
 
     def is_shipping_configured(self):
         # TODO: move to base, or children
-        ret = super()
+        ret = super().is_shipping_configured()
         if ret is False:
             return False
 
+        if self.shipping_auth_mode() == self.SHIPPING_AUTH_CERTIFICATES:
+            if not self.get_ingress_url():
+                logger.error('METRICS_UTILITY_CRC_INGRESS_URL is not set')
+                return False
+            return True
+
         if self.shipping_auth_mode() == self.SHIPPING_AUTH_SERVICE_ACCOUNT:
             if not self.get_ingress_url():
                 logger.error('METRICS_UTILITY_CRC_INGRESS_URL is not set')
@@ -100,6 +162,18 @@ def _send_data(self, url, files, session):
                 timeout=(31, 31),
             )
 
+        elif self.shipping_auth_mode() == self.SHIPPING_AUTH_CERTIFICATES:
+            # session.cert is already set by base ship(); just POST with server cert verification.
+            proxies = {'https': self.get_proxy_url()} if self.get_proxy_url() else {}
+            response = session.post(
+                url,
+                files=files,
+                verify=self.CERT_PATH,
+                proxies=proxies,
+                headers=session.headers,
+                timeout=(31, 31),
+            )
+
         elif self.shipping_auth_mode() == self.SHIPPING_AUTH_USERPASS:
             response = session.post(
                 url,

metrics_utility/library/candlepin/client.py

@@ -0,0 +1,220 @@
+import os
+import tempfile
+import uuid as _uuid_mod
+
+from datetime import datetime, timezone
+
+import requests
+
+from metrics_utility.logger import logger
+
+
+class CandlepinClient:
+    """
+    Minimal Candlepin REST client for certificate lifecycle operations.
+
+    All API calls authenticate with the consumer identity certificate (mTLS),
+    matching the pattern used by subscription-manager after initial registration.
+
+    Candlepin's own CA is not in the system trust store, so TLS server verification
+    defaults to False unless a CA path is provided via ``candlepin_ca``.
+    """
+
+    DEFAULT_CANDLEPIN_URL = 'https://subscription.rhsm.redhat.com/subscription'
+
+    def __init__(self, base_url=None, candlepin_ca=None, proxy=None):
+        self.base_url = (base_url or self.DEFAULT_CANDLEPIN_URL).rstrip('/')
+        self.verify = candlepin_ca if candlepin_ca else False
+        if proxy:
+            # Normalise: strip any existing scheme, then prefix per-protocol.
+            host = proxy.split('://', 1)[-1]
+            self.proxies = {'https': f'https://{host}', 'http': f'http://{host}'}
+        else:
+            self.proxies = {}
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def register_consumer(self, username, password, org, install_uuid=None):
+        """POST /consumers?owner={org} — register a new AAP consumer with basic auth.
+
+        Uses the customer's Red Hat subscription credentials (SUBSCRIPTIONS_USERNAME /
+        SUBSCRIPTIONS_PASSWORD from AWX conf_setting) to register this controller
+        instance as a Candlepin consumer and obtain an identity certificate for mTLS.
+
+        Args:
+            username:     Red Hat subscription username (from SUBSCRIPTIONS_USERNAME).
+            password:     Red Hat subscription password (from SUBSCRIPTIONS_PASSWORD).
+            org:          Candlepin owner/org key (from LICENSE.account_number).
+            install_uuid: AWX INSTALL_UUID used as the consumer's aap.instance_uuid
+                          fact; falls back to a random UUID if not provided.
+
+        Returns:
+            Tuple ``(cert_pem, key_pem, consumer_uuid)``.
+
+        Raises:
+            RuntimeError on any network or API failure.
+        """
+        url = f'{self.base_url}/consumers'
+        instance_uuid = install_uuid or str(_uuid_mod.uuid4())
+        payload = {
+            'name': f'aap-{instance_uuid[:8]}',
+            'type': {'label': 'aap'},
+            'facts': {
+                'system.certificate_version': '3.3',
+                'system.name': 'aap-controller',
+                'aap.instance_uuid': instance_uuid,
+            },
+        }
+        try:
+            resp = requests.post(
+                url,
+                params={'owner': org},
+                auth=(username, password),
+                json=payload,
+                headers={'Content-Type': 'application/json'},
+                verify=self.verify,
+                proxies=self.proxies,
+                timeout=120,
+            )
+        except Exception as e:
+            raise RuntimeError(f'Candlepin register_consumer network error: {e}') from e
+
+        if not resp.ok:
+            raise RuntimeError(f'Candlepin register_consumer failed with status {resp.status_code}: {resp.text}')
+
+        try:
+            body = resp.json()
+            consumer_uuid = body.get('uuid')
+            id_cert = body.get('idCert', {})
+            cert_pem = id_cert.get('cert')
+            key_pem = id_cert.get('key')
+        except Exception as e:
+            raise RuntimeError(f'Candlepin register_consumer: could not parse response JSON: {e}') from e
+
+        if not consumer_uuid or not cert_pem or not key_pem:
+            raise RuntimeError('Candlepin register_consumer: response missing uuid, idCert.cert or idCert.key')
+
+        logger.info(f'Candlepin consumer registered successfully (uuid={consumer_uuid})')
+        return cert_pem, key_pem, consumer_uuid
+
+    def checkin(self, consumer_uuid, cert_pem, key_pem):
+        """PUT /consumers/{uuid} — reset inactivity timer.
+
+        Best-effort: logs a warning on failure but never raises so that a
+        transient Candlepin outage cannot abort a gather run.
+
+        Returns True on success, False on any failure.
+        """
+        url = f'{self.base_url}/consumers/{consumer_uuid}'
+        try:
+            with self._temp_cert_files(cert_pem, key_pem) as (cert_path, key_path):
+                resp = requests.put(
+                    url,
+                    cert=(cert_path, key_path),
+                    json={'facts': {'aap.last_checkin': datetime.now(timezone.utc).isoformat()}},
+                    headers={'Content-Type': 'application/json'},
+                    verify=self.verify,
+                    proxies=self.proxies,
+                    timeout=30,
+                )
+            if resp.status_code in (200, 204):
+                logger.info(f'Candlepin check-in successful for consumer {consumer_uuid}')
+                return True
+            logger.warning(f'Candlepin check-in returned unexpected status {resp.status_code} for consumer {consumer_uuid}')
+            return False
+        except Exception as e:
+            logger.warning(f'Candlepin check-in failed for consumer {consumer_uuid}: {e}')
+            return False
+
+    def regenerate_cert(self, consumer_uuid, cert_pem, key_pem):
+        """POST /consumers/{uuid} — regenerate the identity certificate.
+
+        Returns ``(new_cert_pem, new_key_pem)`` on success.
+        Raises ``RuntimeError`` on API or parsing failure so the caller can
+        decide whether to fall back to service-account auth.
+        """
+        url = f'{self.base_url}/consumers/{consumer_uuid}'
+        with self._temp_cert_files(cert_pem, key_pem) as (cert_path, key_path):
+            try:
+                resp = requests.post(
+                    url,
+                    cert=(cert_path, key_path),
+                    verify=self.verify,
+                    proxies=self.proxies,
+                    timeout=120,
+                )
+            except Exception as e:
+                raise RuntimeError(f'Candlepin regenerate_cert network error for consumer {consumer_uuid}: {e}') from e
+
+        if not resp.ok:
+            raise RuntimeError(f'Candlepin regenerate_cert failed with status {resp.status_code} for consumer {consumer_uuid}: {resp.text}')
+
+        try:
+            body = resp.json()
+            id_cert = body.get('idCert', {})
+            new_cert_pem = id_cert.get('cert')
+            new_key_pem = id_cert.get('key')
+        except Exception as e:
+            raise RuntimeError(f'Candlepin regenerate_cert: could not parse response JSON: {e}') from e
+
+        if not new_cert_pem or not new_key_pem:
+            raise RuntimeError(f'Candlepin regenerate_cert: response did not contain idCert.cert / idCert.key for consumer {consumer_uuid}')
+
+        logger.info(f'Candlepin cert regenerated successfully for consumer {consumer_uuid}')
+        return new_cert_pem, new_key_pem
+
+    # ------------------------------------------------------------------
+    # Private helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _write_temp_pem(content):
+        """Write PEM content to a secure temp file. Returns (fd, path)."""
+        fd, path = tempfile.mkstemp(prefix='candlepin_', suffix='.pem')
+        try:
+            os.chmod(path, 0o600)
+            with os.fdopen(fd, 'w') as f:
+                f.write(content)
+            fd = None
+        except Exception:
+            if fd is not None:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+            if os.path.exists(path):
+                os.unlink(path)
+            raise
+        return path
+
+    @staticmethod
+    def _unlink_safe(path):
+        try:
+            if path and os.path.exists(path):
+                os.unlink(path)
+        except OSError as e:
+            logger.warning(f'Could not remove temp cert file {path}: {e}')
+
+    class _temp_cert_files:
+        """Context manager: writes cert + key to secure temp files, cleans up on exit."""
+
+        def __init__(self, cert_pem, key_pem):
+            self._cert_pem = cert_pem
+            self._key_pem = key_pem
+            self._cert_path = None
+            self._key_path = None
+
+        def __enter__(self):
+            self._cert_path = CandlepinClient._write_temp_pem(self._cert_pem)
+            try:
+                self._key_path = CandlepinClient._write_temp_pem(self._key_pem)
+            except Exception:
+                CandlepinClient._unlink_safe(self._cert_path)
+                raise
+            return self._cert_path, self._key_path
+
+        def __exit__(self, *_):
+            CandlepinClient._unlink_safe(self._cert_path)
+            CandlepinClient._unlink_safe(self._key_path)