You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@skywalking.apache.org by "songzhendong (via GitHub)" <gi...@apache.org> on 2026/09/12 14:43:47 UTC

[PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

songzhendong opened a new pull request, #419:
URL: https://github.com/apache/skywalking-python/pull/419

## Summary

Adds collector **TLS / mTLS** for the Python agent's **gRPC and HTTP** reporters
via shared helpers in `skywalking/utils/tls.py`.

- TLS is enabled by `SW_AGENT_FORCE_TLS` and/or a **readable + parseable** CA PEM
(`SW_AGENT_SSL_TRUSTED_CA_PATH`).
- mTLS requires a usable CA **and** both client cert/key PEMs
(`SW_AGENT_SSL_CERT_CHAIN_PATH` / `SW_AGENT_SSL_KEY_PATH`).
- PKCS#1 private keys are normalized to PKCS#8; HTTP mTLS temp PEMs are
fork-safe (`os.register_at_fork`).
- Symlinks to regular files are followed (Kubernetes secret mounts).
- gRPC does **not** set `grpc.ssl_target_name_override`. With
`grpc.default_authority` set to the first configured endpoint, TLS peer-name
checks follow that authority (not the shuffled multi-address dial target).

Kafka protocol is out of scope for these options.

## Security / hardening fixes (review follow-ups)

These close concrete gaps against the documented contract
(“TLS misconfiguration must not abort host process start” / correct degrade table):

1. **Path resolution must not block HTTP agent start**
`ssl_file_path()` now wraps `Path.expanduser()` + `resolve(strict=True)` and
catches `OSError` / `ValueError` / `RuntimeError` (Windows permission errors,
symlink loops, unresolvable `~user`, etc.).
`collector_http_scheme()` (used from HTTP sync/aio reporter `__init__`) also
has a top-level fallback so scheme selection cannot raise into bootstrap.

2. **CA existence vs load TOCTOU (incl. K8s secret rotation)**
TLS enablement is based on the **current** successful CA load (`ca_usable`),
not a stale “file exists” probe alone. Without `FORCE_TLS`, a CA that vanishes
or fails mid-load stays **plaintext** (does not silently flip to HTTPS +
system trust). HTTP aio loads CA from in-memory `cadata`; sync HTTP can
persist loaded CA bytes to a temp PEM if the path races away.

3. **FORCE_TLS credential failure is fail-open (documented)**
If gRPC/process-trust credentials or `SSLContext` construction ultimately
fails, reporters **warn and stay plaintext** rather than aborting start.
This is intentional best-effort encryption, not fail-closed — spelled out in
`docs/en/setup/Intrusive.md` / `Configuration.md`.

4. **Diagnostics: mTLS warning without usable CA**
Configuring client cert/key without a usable CA now logs the dedicated mTLS
warning even when `FORCE_TLS` is off (previously only covered with
`FORCE_TLS=true`).

5. **Docs aligned with C-core authority / TLS peer-name**
Documented that `:authority` and TLS peer-name checks use
`grpc.default_authority` (first configured endpoint), not the encoded
multi-backend channel target used for dialing.

6. **mTLS e2e healthcheck**
Sharing-server mTLS e2e probes port **11811** (not core 11800).

## Test plan

- [x] Unit: `tests/unit/test_tls.py`, `tests/unit/test_grpc_channel.py`
(path errors, TOCTOU→plaintext, mTLS-without-CA warning, degrade matrix,
PKCS#1, fork temp safety)
- [x] Real-OAP e2e: `tests/e2e/case/grpc/ssl/`, `tests/e2e/case/grpc/mtls/`
- [x] Fork CI: https://github.com/songzhendong/skywalking-python/actions/runs/34698689614
- [ ] Local HTTP mTLS against a TLS terminator that requires a client cert
(OAP HTTP/REST TLS is **server-side only** — no real-OAP HTTP mTLS e2e)

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.

wu-sheng merged PR #419:
URL: https://github.com/apache/skywalking-python/pull/419


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "songzhendong (via GitHub)" <gi...@apache.org>.

songzhendong commented on code in PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#discussion_r3996808514


##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,678 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return data
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    body = text.replace(_PKCS1_PEM_HEADER, '').replace(_PKCS1_PEM_FOOTER, '')
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    When custom CA bytes were loaded but the CA path is no longer readable (symlink
+    race), verify uses a process-local temp PEM of those bytes instead of silently
+    falling back to the system trust store.
+
+    Temp-file failures drop client certs only (one-way TLS), with a warning.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    if root_certificates is not None:
+        if ca_path is not None:
+            verify: object = str(ca_path)

Review Comment:
   Fixed in `3d3d544`: custom CA verify now always uses a process-lifetime temp PEM snapshot, not the resolved K8s `..data` target. Covered by `test_requests_verify_survives_k8s_style_ca_rotation`.



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,678 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return data
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    body = text.replace(_PKCS1_PEM_HEADER, '').replace(_PKCS1_PEM_FOOTER, '')
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body)

Review Comment:
   Fixed in `3d3d544`: decode only the bytes between matching PKCS#1 BEGIN/END delimiters. Covered by `test_normalize_pkcs1_ignores_preamble_outside_pem_block`.



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,678 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return data
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    body = text.replace(_PKCS1_PEM_HEADER, '').replace(_PKCS1_PEM_FOOTER, '')
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    When custom CA bytes were loaded but the CA path is no longer readable (symlink
+    race), verify uses a process-local temp PEM of those bytes instead of silently
+    falling back to the system trust store.
+
+    Temp-file failures drop client certs only (one-way TLS), with a warning.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    if root_certificates is not None:
+        if ca_path is not None:
+            verify: object = str(ca_path)
+        else:
+            # Prefer in-memory CA over system trust when the path raced away.
+            try:
+                verify = _pem_bytes_to_temp_file(root_certificates, '.crt')
+            except OSError as exc:
+                _warn_once(
+                    f'requests-ca-temp:{exc}',
+                    'Failed to persist trusted CA temp file (%s); '
+                    'using process trust store for HTTP verify.',
+                    exc,
+                )
+                verify = True
+    else:
+        verify = True
+
+    if private_key is None or certificate_chain is None:
+        return verify, None
+
+    pair = _mtls_cert_key_files(certificate_chain, private_key)
+    return verify, pair
+
+
+def configure_requests_session(session) -> None:
+    try:
+        verify, cert = requests_tls_settings()
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'requests-tls:{exc}',
+            'Failed to configure HTTP reporter TLS (%s); falling back.',
+            exc,
+        )
+        if config.agent_force_tls or ssl_file_path(config.agent_ssl_trusted_ca_path):
+            # Prefer system-trust https over aborting agent start.
+            session.verify = True
+            session.cert = None
+        return
+    session.verify = verify
+    if cert is not None:
+        session.cert = cert
+
+
+def ssl_context_for_collector() -> Optional[ssl.SSLContext]:
+    """
+    stdlib SSLContext for aiohttp, or None when the collector stays plaintext.
+
+    Degrade on SSLError / OSError (never raise into agent bootstrap):
+    - bad / unparsable custom CA + FORCE_TLS → process trust store, no client cert
+    - bad / unparsable custom CA without FORCE_TLS → plaintext (None)
+    - bad client cert/key or temp-file failure → one-way TLS (CA or system trust)
+
+    Custom CA is loaded from in-memory PEM bytes (``cadata``) when present so a
+    path race after ``tls_pem_material`` cannot silently switch to system trust.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+
+    def _system_trust_context() -> ssl.SSLContext:
+        return ssl.create_default_context()
+
+    ctx: Optional[ssl.SSLContext]
+    if root_certificates is not None:
+        try:
+            # cadata avoids re-reading the CA path (TOCTOU with K8s secret mounts).
+            ctx = ssl.create_default_context(
+                cadata=root_certificates.decode('ascii', errors='strict'),

Review Comment:
   Fixed in `3d3d544`: `_load_trusted_ca` extracts ASCII CERTIFICATE blocks so BOM/UTF-8 preamble stay usable via `cadata`. Covered by `test_aio_context_accepts_bom_and_utf8_preamble_ca`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "songzhendong (via GitHub)" <gi...@apache.org>.

songzhendong commented on PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#issuecomment-5647108601

Thanks for the detailed P2 reports and repros. Addressed in `3d3d544`:

**P2-1 — HTTP CA path after Kubernetes secret rotation**
`requests_tls_settings()` no longer stores the resolved symlink target in `session.verify`. Custom CA now always uses a process-lifetime temp snapshot of the validated PEM bytes (same pattern as client mTLS material), so rotating `..data` / removing the old version directory cannot invalidate an already-configured session.

**P2-2 — PKCS#1 preamble outside PEM delimiters**
`normalize_private_key_pem()` now base64-decodes only the content between matching `BEGIN RSA PRIVATE KEY` / `END RSA PRIVATE KEY` delimiters. Comments or `openssl rsa -text` preamble outside the block no longer corrupt the key or disable mTLS.

**P2-3 — aio SSLContext rejecting CA that `cafile` accepts**
`_load_trusted_ca()` extracts/normalizes `CERTIFICATE` PEM blocks (strips UTF-8 BOM and non-ASCII preamble) so the same bytes work for both validation and `ssl.create_default_context(cadata=...)`. BOM / UTF-8 comment CAs keep `https://` with a custom async context.

Covered by unit tests aligned with your repros (`test_requests_verify_survives_k8s_style_ca_rotation`, `test_normalize_pkcs1_ignores_preamble_outside_pem_block`, `test_aio_context_accepts_bom_and_utf8_preamble_ca`). Please take another look.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "songzhendong (via GitHub)" <gi...@apache.org>.

songzhendong commented on PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#issuecomment-5691287280

Thanks for the review — both P2s are valid on the previous PR head (`1327f8b`).

They are fixed in `3b3906d` (now on this PR):

1. **HTTP CA fallback + K8s symlink rotation** — when the CA temp snapshot cannot be written, fall back to the configured CA path **without** resolving symlinks (`_configured_ca_path_for_verify`), so `session.verify` is not pinned to a deleted `v1/` target after Secret rotation.
2. **Mixed TRUSTED + CERTIFICATE bundle** — if the CA PEM contains `TRUSTED CERTIFICATE`, always load via `cafile` (never `cadata`), so trusted blocks are not silently dropped from a mixed bundle.

Separately, while re-checking we also found and fixed a few non-blocking polish items (`ca6d3be`):

3. **Orphan mTLS temp cert** — if key `mkstemp` fails after the cert temp was written, unlink the orphaned `.crt` immediately instead of leaving it until process exit.
4. **HTTP scheme / TLS settings TOCTOU** — sync/aio HTTP reporters now share one `tls_pem_material()` load for URL scheme and session/`SSLContext`, so the two cannot disagree across a race.
5. **Broad `except` in `configure_requests_session`** — on unexpected failure, prefer the configured CA path for `session.verify` rather than dropping straight to system trust.
6. **Dead code** — removed unused `_extract_pem_blocks` (only `_extract_ca_pem_blocks` remains).
7. **CA hot-reload** — left intentional: process-lifetime CA snapshot + unresolved-symlink fallback for K8s rotation; agent restart still required to pick up in-place CA byte changes.

Pushed in `3b3906d..ca6d3be` — please re-check.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "songzhendong (via GitHub)" <gi...@apache.org>.

songzhendong commented on PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#issuecomment-5650716304

Thanks for the follow-up review. The two new findings are addressed in `1327f8b`:

**P2 — Preserve the readable custom CA when its snapshot cannot be written**
If creating the process-lifetime CA temp snapshot raises `OSError` (e.g. read-only temp dir), `requests_tls_settings()` now falls back to the still-readable configured CA path instead of `verify=True` (Requests' default trust store). A usable private CA is no longer discarded merely because a copy cannot be written. Snapshot remains the preferred path so Kubernetes `..data` rotation stays covered when temp is writable.

**P2 — Preserve OpenSSL trusted-certificate blocks for synchronous HTTP**
`_load_trusted_ca()` now extracts both `CERTIFICATE` and `TRUSTED CERTIFICATE` blocks, keeping their PEM labels and trust attributes (no relabeling to `CERTIFICATE`). Sync HTTP snapshots therefore keep `openssl x509 -trustout` CAs. For aio, `cadata` still cannot load `TRUSTED CERTIFICATE`; we fall back to `cafile` via the same snapshot (or the configured path) so the custom CA is not dropped.

Covered by `test_requests_ca_temp_failure_falls_back_to_configured_path`, `test_extract_ca_pem_blocks_preserves_trusted_label`, and `test_trusted_certificate_pem_enables_sync_http_tls`. Please take another look.

--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.

wu-sheng commented on code in PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#discussion_r3996602523


##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,678 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return data
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    body = text.replace(_PKCS1_PEM_HEADER, '').replace(_PKCS1_PEM_FOOTER, '')
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body)

Review Comment:
   **[P2] Decode only the content inside the RSA PEM delimiters**
   
   A valid unencrypted PKCS#1 key can contain comments or metadata before its PEM block. OpenSSL accepts such files, but removing the delimiters from the entire file leaves that surrounding text in `body`. `b64decode()` incorporates Base64 characters from the preamble into the key, or raises for invalid padding. The resulting key fails validation, so the agent drops the client identity and cannot report to a collector that requires mTLS. Besides the short comment below, this is reproducible with the output of `openssl rsa -traditional -text -in key.pem -out keytext.pem`.
   
   Please extract only the content between the matching `BEGIN RSA PRIVATE KEY` and `END RSA PRIVATE KEY` delimiters before decoding, and cover a valid preamble in the regression test.
   
   Run from the PR checkout with the project dependencies installed (the snippets reuse the PEM fixtures already in this PR):
   
   ```bash
   poetry run python - <<'PY'
   import ssl
   from pathlib import Path
   from tempfile import TemporaryDirectory
   from skywalking import config
   from skywalking.utils import tls
   from tests.unit.test_tls import _TEST_CA_CERT, _TEST_CLIENT_CERT, _TEST_CLIENT_KEY_PKCS1
   
   with TemporaryDirectory() as tmp:
       ca, cert, key = (Path(tmp) / name for name in ('ca.pem', 'cert.pem', 'key.pem'))
       ca.write_bytes(_TEST_CA_CERT)
       cert.write_bytes(_TEST_CLIENT_CERT)
       key.write_bytes(b'# This is a private key\n' + _TEST_CLIENT_KEY_PKCS1)
       ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
       ctx.load_cert_chain(cert, key)
       print('Original PKCS#1 key: loaded successfully')
   
       normalized = Path(tmp) / 'normalized.pem'
       normalized.write_bytes(tls.normalize_private_key_pem(key.read_bytes()))
       try:
           ctx.load_cert_chain(cert, normalized)
       except ssl.SSLError as exc:
           print('Normalized key:', type(exc).__name__)
   
       config.agent_force_tls = False
       config.agent_ssl_trusted_ca_path = str(ca)
       config.agent_ssl_cert_chain_path = str(cert)
       config.agent_ssl_key_path = str(key)
       roots, private_key, chain = tls.tls_pem_material()
       print('Agent TLS:', roots is not None)
       print('Agent mTLS:', private_key is not None and chain is not None)
   PY
   ```
   
   Observed output, plus the warning about failing to enable mTLS:
   
   ```text
   Original PKCS#1 key: loaded successfully
   Normalized key: SSLError
   Agent TLS: True
   Agent mTLS: False
   ```
   
   Expected: adding text outside the PEM block does not corrupt an otherwise loadable key or disable the client identity. This reproduction checks parsing/key matching locally; the fixture certificates are not being used to claim a successful mTLS handshake.
   



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,678 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return data
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    body = text.replace(_PKCS1_PEM_HEADER, '').replace(_PKCS1_PEM_FOOTER, '')
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    When custom CA bytes were loaded but the CA path is no longer readable (symlink
+    race), verify uses a process-local temp PEM of those bytes instead of silently
+    falling back to the system trust store.
+
+    Temp-file failures drop client certs only (one-way TLS), with a warning.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    if root_certificates is not None:
+        if ca_path is not None:
+            verify: object = str(ca_path)
+        else:
+            # Prefer in-memory CA over system trust when the path raced away.
+            try:
+                verify = _pem_bytes_to_temp_file(root_certificates, '.crt')
+            except OSError as exc:
+                _warn_once(
+                    f'requests-ca-temp:{exc}',
+                    'Failed to persist trusted CA temp file (%s); '
+                    'using process trust store for HTTP verify.',
+                    exc,
+                )
+                verify = True
+    else:
+        verify = True
+
+    if private_key is None or certificate_chain is None:
+        return verify, None
+
+    pair = _mtls_cert_key_files(certificate_chain, private_key)
+    return verify, pair
+
+
+def configure_requests_session(session) -> None:
+    try:
+        verify, cert = requests_tls_settings()
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'requests-tls:{exc}',
+            'Failed to configure HTTP reporter TLS (%s); falling back.',
+            exc,
+        )
+        if config.agent_force_tls or ssl_file_path(config.agent_ssl_trusted_ca_path):
+            # Prefer system-trust https over aborting agent start.
+            session.verify = True
+            session.cert = None
+        return
+    session.verify = verify
+    if cert is not None:
+        session.cert = cert
+
+
+def ssl_context_for_collector() -> Optional[ssl.SSLContext]:
+    """
+    stdlib SSLContext for aiohttp, or None when the collector stays plaintext.
+
+    Degrade on SSLError / OSError (never raise into agent bootstrap):
+    - bad / unparsable custom CA + FORCE_TLS → process trust store, no client cert
+    - bad / unparsable custom CA without FORCE_TLS → plaintext (None)
+    - bad client cert/key or temp-file failure → one-way TLS (CA or system trust)
+
+    Custom CA is loaded from in-memory PEM bytes (``cadata``) when present so a
+    path race after ``tls_pem_material`` cannot silently switch to system trust.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+
+    def _system_trust_context() -> ssl.SSLContext:
+        return ssl.create_default_context()
+
+    ctx: Optional[ssl.SSLContext]
+    if root_certificates is not None:
+        try:
+            # cadata avoids re-reading the CA path (TOCTOU with K8s secret mounts).
+            ctx = ssl.create_default_context(
+                cadata=root_certificates.decode('ascii', errors='strict'),

Review Comment:
   **[P2] Preserve accepted CA PEMs when constructing the aiohttp context**
   
   `_load_trusted_ca()` validates the file through OpenSSL's `cafile` loader, which accepts a UTF-8 BOM or a non-ASCII preamble outside the PEM block. The strict ASCII decoding here then rejects the same validated file. With `agent_force_tls=False`, `collector_http_scheme()` has already selected `https://`, but `ssl_context_for_collector()` returns `None`. `_aiohttp_session()` consequently creates a default session, losing the custom CA and any client identity while continuing to use HTTPS. In a local mTLS check, a CA with a UTF-8 comment worked through the synchronous HTTP reporter but the async reporter raised `ClientConnectorCertificateError`.
   
   Please use consistent PEM parsing for validation and context construction, for example extracting/normalizing the certificate PEM blocks or loading a stable snapshot through a compatible loader. Changing only the URL to plaintext would still discard a valid configured CA.
   
   Run from the PR checkout with the project dependencies installed (the snippets reuse the PEM fixtures already in this PR):
   
   ```bash
   poetry run python - <<'PY'
   import ssl
   import tempfile
   from pathlib import Path
   from skywalking import config
   from skywalking.utils.tls import collector_http_scheme, ssl_context_for_collector
   from tests.unit.test_tls import _TEST_CA_CERT
   
   config.agent_force_tls = False
   config.agent_ssl_cert_chain_path = ''
   config.agent_ssl_key_path = ''
   with tempfile.TemporaryDirectory() as tmp:
       ca = Path(tmp) / 'ca.crt'
       config.agent_ssl_trusted_ca_path = str(ca)
       for label, prefix in [('plain PEM', b''), ('UTF-8 BOM', b'\xef\xbb\xbf'),
                             ('UTF-8 comment', '# 注释\n'.encode('utf-8'))]:
           ca.write_bytes(prefix + _TEST_CA_CERT)
           ssl.create_default_context(cafile=str(ca))
           print(label, '| cafile accepted |', collector_http_scheme(),
                 '| custom context:', ssl_context_for_collector() is not None)
   PY
   ```
   
   Observed output, plus ASCII-decoding warnings for the last two cases:
   
   ```text
   plain PEM | cafile accepted | https:// | custom context: True
   UTF-8 BOM | cafile accepted | https:// | custom context: False
   UTF-8 comment | cafile accepted | https:// | custom context: False
   ```
   
   Expected: all three OpenSSL-accepted representations preserve the configured CA in the async SSL context.
   



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,678 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return data
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    body = text.replace(_PKCS1_PEM_HEADER, '').replace(_PKCS1_PEM_FOOTER, '')
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    When custom CA bytes were loaded but the CA path is no longer readable (symlink
+    race), verify uses a process-local temp PEM of those bytes instead of silently
+    falling back to the system trust store.
+
+    Temp-file failures drop client certs only (one-way TLS), with a warning.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    if root_certificates is not None:
+        if ca_path is not None:
+            verify: object = str(ca_path)

Review Comment:
   **[P2] Keep the HTTP CA path usable after Kubernetes secret rotation**
   
   `ssl_file_path()` resolves the configured symlink, and this branch stores that resolved target in `session.verify` for the lifetime of the reporter. With a Kubernetes-style mount (`ca.crt -> ..data/ca.crt`), a normal secret update switches `..data` and removes the old version directory. The configured CA remains readable, but every subsequent Requests call uses the deleted old path and raises `OSError`. This happens even when the CA bytes are identical before and after rotation; it does not require certificate renewal or a startup race. In a local mTLS check, the same session returned HTTP 200 before rotation and failed with the invalid-CA-path error afterward.
   
   Please retain a process-lifetime snapshot of the validated CA bytes, as is already done for client material, or otherwise ensure the session follows a usable stable path. The current temp-file fallback only covers a path disappearing during session construction.
   
   Run from the PR checkout with the project dependencies installed (the snippets reuse the PEM fixtures already in this PR):
   
   ```bash
   poetry run python - <<'PY'
   import os
   import shutil
   import tempfile
   from pathlib import Path
   import requests
   from skywalking import config
   from skywalking.utils.tls import configure_requests_session
   from tests.unit.test_tls import _TEST_CA_CERT
   
   config.agent_force_tls = False
   config.agent_ssl_cert_chain_path = ''
   config.agent_ssl_key_path = ''
   with tempfile.TemporaryDirectory() as tmp, requests.Session() as session:
       root = Path(tmp)
       for version in ('v1', 'v2'):
           (root / version).mkdir()
           (root / version / 'ca.crt').write_bytes(_TEST_CA_CERT)
       (root / '..data').symlink_to('v1')
       ca = root / 'ca.crt'
       ca.symlink_to('..data/ca.crt')
       config.agent_ssl_trusted_ca_path = str(ca)
       configure_requests_session(session)
       session.trust_env = False
       print('Stored CA exists before rotation:', Path(session.verify).is_file())
   
       (root / '..data-next').symlink_to('v2')
       os.replace(root / '..data-next', root / '..data')
       shutil.rmtree(root / 'v1')
       print('Configured CA remains valid:', ca.read_bytes() == _TEST_CA_CERT)
       print('Stored CA exists after rotation:', Path(session.verify).is_file())
       try:
           session.post('https://127.0.0.1:1/v3/segment', json={}, timeout=1)
       except OSError as exc:
           print(type(exc).__name__ + ':', exc)
   PY
   ```
   
   Observed output (temporary directory varies):
   
   ```text
   Stored CA exists before rotation: True
   Configured CA remains valid: True
   Stored CA exists after rotation: False
   OSError: Could not find a suitable TLS CA certificate bundle, invalid path: .../v1/ca.crt
   ```
   
   No collector is needed for this minimal reproduction: the loopback request fails during CA path validation, before connecting. Expected: the session retains usable CA material and does not fail because the old secret directory was removed.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.

wu-sheng commented on code in PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#discussion_r4021344582


##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,806 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+
+    Returns ASCII PEM with ``CERTIFICATE`` and/or ``TRUSTED CERTIFICATE`` blocks
+    extracted (labels and trust attributes preserved). UTF-8 BOM / preamble
+    accepted by ``cafile`` stay usable for ``cadata`` and temp verify files.
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return _extract_ca_pem_blocks(data)
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+
+    Only the bytes between the matching PKCS#1 BEGIN/END delimiters are
+    decoded — comments or ``openssl rsa -text`` preamble outside the block
+    must not corrupt the key (OpenSSL accepts such files).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    start = text.find(_PKCS1_PEM_HEADER)
+    end = text.find(_PKCS1_PEM_FOOTER, start)
+    if start < 0 or end < 0:
+        raise ValueError('Invalid PKCS#1 private key PEM: missing BEGIN/END delimiters')
+    body = text[start + len(_PKCS1_PEM_HEADER):end]
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body, validate=False)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _extract_pem_blocks(data: bytes, label: str) -> bytes:
+    """
+    Return ASCII PEM containing only ``BEGIN/END {label}`` blocks.
+
+    Strips UTF-8 BOM and ignores preamble/comments outside PEM delimiters so
+    OpenSSL-accepted files (BOM, UTF-8 comments) stay usable for ``cadata`` and
+    temp-file verify paths that require clean ASCII PEM.
+    """
+    text = data.decode('utf-8', errors='ignore').lstrip('\ufeff')
+    header = f'-----BEGIN {label}-----'
+    footer = f'-----END {label}-----'
+    blocks: List[str] = []
+    pos = 0
+    while True:
+        start = text.find(header, pos)
+        if start < 0:
+            break
+        end = text.find(footer, start)
+        if end < 0:
+            break
+        end += len(footer)
+        blocks.append(text[start:end].strip() + '\n')
+        pos = end
+    if not blocks:
+        raise ValueError(f'No {label} PEM block found')
+    return ''.join(blocks).encode('ascii')
+
+
+# OpenSSL ``openssl x509 -trustout`` emits TRUSTED CERTIFICATE (with trust
+# auxiliary). Keep that label — do not relabel as CERTIFICATE.
+_CA_PEM_LABELS = ('TRUSTED CERTIFICATE', 'CERTIFICATE')
+
+
+def _extract_ca_pem_blocks(data: bytes) -> bytes:
+    """
+    Return ASCII PEM with ``CERTIFICATE`` / ``TRUSTED CERTIFICATE`` blocks only.
+
+    Preserves each block's PEM label and trust attributes. Strips UTF-8 BOM and
+    ignores preamble outside delimiters. ``TRUSTED CERTIFICATE`` is searched
+    first so its header is not confused with plain ``CERTIFICATE``.
+    """
+    text = data.decode('utf-8', errors='ignore').lstrip('\ufeff')
+    found: List[Tuple[int, str]] = []
+    for label in _CA_PEM_LABELS:
+        header = f'-----BEGIN {label}-----'
+        footer = f'-----END {label}-----'
+        pos = 0
+        while True:
+            start = text.find(header, pos)
+            if start < 0:
+                break
+            end = text.find(footer, start)
+            if end < 0:
+                break
+            end += len(footer)
+            found.append((start, text[start:end].strip() + '\n'))
+            pos = end
+    if not found:
+        raise ValueError('No CERTIFICATE or TRUSTED CERTIFICATE PEM block found')
+    found.sort(key=lambda item: item[0])
+    return ''.join(block for _, block in found).encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS / CA temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_ca_file_cache_key: Optional[bytes] = None
+_ca_file_cache: Optional[str] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    global _ca_file_cache, _ca_file_cache_key, _mtls_file_cache, _mtls_file_cache_key
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    global _ca_file_cache, _ca_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def _ca_verify_temp_file(root_certificates: bytes) -> str:
+    """Process-lifetime temp path for trusted CA bytes (survives K8s secret rotation)."""
+    global _ca_file_cache_key, _ca_file_cache
+    if (
+        _ca_file_cache is not None
+        and _ca_file_cache_key == root_certificates
+        and os.path.exists(_ca_file_cache)
+    ):
+        return _ca_file_cache
+    path = _pem_bytes_to_temp_file(root_certificates, '.crt')
+    _ca_file_cache_key = root_certificates
+    _ca_file_cache = path
+    return path
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    Custom CA prefers a process-lifetime temp snapshot of the validated PEM bytes
+    (never the resolved symlink target), so Kubernetes secret rotation that
+    removes the old ``..data`` version cannot invalidate an already-configured
+    session. If the snapshot cannot be written (e.g. read-only temp dir), fall
+    back to the still-readable configured CA path rather than discarding the
+    private CA for Requests' default trust store.
+
+    Client-cert temp-file failures drop mTLS only (one-way TLS), with a warning.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    if root_certificates is not None:
+        try:
+            verify: object = _ca_verify_temp_file(root_certificates)
+        except OSError as exc:
+            ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+            if ca_path is not None:
+                _warn_once(
+                    f'requests-ca-temp:{exc}',
+                    'Failed to persist trusted CA temp file (%s); '
+                    'falling back to configured CA path for HTTP verify.',
+                    exc,
+                )
+                verify = str(ca_path)

Review Comment:
   **[P2] Preserve the configured symlink in the no-snapshot fallback**
   
   `ssl_file_path()` resolves symlinks, so this fallback stores the versioned `v1/ca.crt` target instead of the configured `ca.crt -> ..data/ca.crt` path. When temporary storage is unavailable, a subsequent Kubernetes secret update still invalidates the existing Requests session: replacing `..data` and deleting `v1` leaves the configured CA readable but makes `session.verify` point to a deleted file. The no-temp and rotation cases need to work together.
   
   I verified this with a local HTTPS server and an actual non-writable temp directory, without mocking file creation: the session returned HTTP 200 before rotation, then raised the invalid-CA-path `OSError` afterward. Assigning the unresolved configured symlink to `session.verify` restored HTTP 200.
   
   Please validate the configured CA but retain its absolute, expanded path **without resolving its symlinks** in this fallback. The normal writable-temp snapshot path can remain unchanged.
   
   Minimal reproduction from the PR checkout with dependencies installed, as a non-root user on Linux/macOS. No collector is required: Requests rejects the missing CA path before connecting.
   
   ```bash
   poetry run python - <<'PY'
   import os
   import shutil
   import tempfile
   from pathlib import Path
   import requests
   from skywalking import config
   from skywalking.utils.tls import configure_requests_session
   from tests.unit.test_tls import _TEST_CA_CERT
   
   config.agent_force_tls = False
   config.agent_ssl_cert_chain_path = ''
   config.agent_ssl_key_path = ''
   with tempfile.TemporaryDirectory() as tmp, requests.Session() as session:
       root = Path(tmp)
       for version in ('v1', 'v2'):
           (root / version).mkdir()
           (root / version / 'ca.crt').write_bytes(_TEST_CA_CERT)
       (root / '..data').symlink_to('v1')
       ca = root / 'ca.crt'
       ca.symlink_to('..data/ca.crt')
       config.agent_ssl_trusted_ca_path = str(ca)
       blocked = root / 'readonly'
       blocked.mkdir()
       blocked.chmod(0o555)
       saved_tempdir = tempfile.tempdir
       try:
           if os.access(blocked, os.W_OK):
               raise SystemExit('Run as a non-root user on a POSIX filesystem.')
           tempfile.tempdir = str(blocked)
           session.trust_env = False
           configure_requests_session(session)
           print('Fallback uses v1:', Path(session.verify).parent.name == 'v1')
           (root / '..data-next').symlink_to('v2')
           os.replace(root / '..data-next', root / '..data')
           shutil.rmtree(root / 'v1')
           print('Configured CA remains valid:', ca.read_bytes() == _TEST_CA_CERT)
           print('Stored verify path exists:', Path(session.verify).exists())
           try:
               session.post('https://127.0.0.1:1/v3/segment', json={}, timeout=1)
           except OSError as exc:
               print(type(exc).__name__ + ':', exc)
       finally:
           tempfile.tempdir = saved_tempdir
           blocked.chmod(0o755)
   PY
   ```
   
   Observed output, plus the expected warning that snapshot creation failed:
   
   ```text
   Fallback uses v1: True
   Configured CA remains valid: True
   Stored verify path exists: False
   OSError: Could not find a suitable TLS CA certificate bundle, invalid path: .../v1/ca.crt
   ```
   
   Expected: after rotation, the fallback still points through the readable configured CA symlink and does not fail because the old version directory was removed.
   



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,806 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+
+    Returns ASCII PEM with ``CERTIFICATE`` and/or ``TRUSTED CERTIFICATE`` blocks
+    extracted (labels and trust attributes preserved). UTF-8 BOM / preamble
+    accepted by ``cafile`` stay usable for ``cadata`` and temp verify files.
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return _extract_ca_pem_blocks(data)
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+
+    Only the bytes between the matching PKCS#1 BEGIN/END delimiters are
+    decoded — comments or ``openssl rsa -text`` preamble outside the block
+    must not corrupt the key (OpenSSL accepts such files).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    start = text.find(_PKCS1_PEM_HEADER)
+    end = text.find(_PKCS1_PEM_FOOTER, start)
+    if start < 0 or end < 0:
+        raise ValueError('Invalid PKCS#1 private key PEM: missing BEGIN/END delimiters')
+    body = text[start + len(_PKCS1_PEM_HEADER):end]
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body, validate=False)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _extract_pem_blocks(data: bytes, label: str) -> bytes:
+    """
+    Return ASCII PEM containing only ``BEGIN/END {label}`` blocks.
+
+    Strips UTF-8 BOM and ignores preamble/comments outside PEM delimiters so
+    OpenSSL-accepted files (BOM, UTF-8 comments) stay usable for ``cadata`` and
+    temp-file verify paths that require clean ASCII PEM.
+    """
+    text = data.decode('utf-8', errors='ignore').lstrip('\ufeff')
+    header = f'-----BEGIN {label}-----'
+    footer = f'-----END {label}-----'
+    blocks: List[str] = []
+    pos = 0
+    while True:
+        start = text.find(header, pos)
+        if start < 0:
+            break
+        end = text.find(footer, start)
+        if end < 0:
+            break
+        end += len(footer)
+        blocks.append(text[start:end].strip() + '\n')
+        pos = end
+    if not blocks:
+        raise ValueError(f'No {label} PEM block found')
+    return ''.join(blocks).encode('ascii')
+
+
+# OpenSSL ``openssl x509 -trustout`` emits TRUSTED CERTIFICATE (with trust
+# auxiliary). Keep that label — do not relabel as CERTIFICATE.
+_CA_PEM_LABELS = ('TRUSTED CERTIFICATE', 'CERTIFICATE')
+
+
+def _extract_ca_pem_blocks(data: bytes) -> bytes:
+    """
+    Return ASCII PEM with ``CERTIFICATE`` / ``TRUSTED CERTIFICATE`` blocks only.
+
+    Preserves each block's PEM label and trust attributes. Strips UTF-8 BOM and
+    ignores preamble outside delimiters. ``TRUSTED CERTIFICATE`` is searched
+    first so its header is not confused with plain ``CERTIFICATE``.
+    """
+    text = data.decode('utf-8', errors='ignore').lstrip('\ufeff')
+    found: List[Tuple[int, str]] = []
+    for label in _CA_PEM_LABELS:
+        header = f'-----BEGIN {label}-----'
+        footer = f'-----END {label}-----'
+        pos = 0
+        while True:
+            start = text.find(header, pos)
+            if start < 0:
+                break
+            end = text.find(footer, start)
+            if end < 0:
+                break
+            end += len(footer)
+            found.append((start, text[start:end].strip() + '\n'))
+            pos = end
+    if not found:
+        raise ValueError('No CERTIFICATE or TRUSTED CERTIFICATE PEM block found')
+    found.sort(key=lambda item: item[0])
+    return ''.join(block for _, block in found).encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS / CA temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_ca_file_cache_key: Optional[bytes] = None
+_ca_file_cache: Optional[str] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    global _ca_file_cache, _ca_file_cache_key, _mtls_file_cache, _mtls_file_cache_key
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    global _ca_file_cache, _ca_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def _ca_verify_temp_file(root_certificates: bytes) -> str:
+    """Process-lifetime temp path for trusted CA bytes (survives K8s secret rotation)."""
+    global _ca_file_cache_key, _ca_file_cache
+    if (
+        _ca_file_cache is not None
+        and _ca_file_cache_key == root_certificates
+        and os.path.exists(_ca_file_cache)
+    ):
+        return _ca_file_cache
+    path = _pem_bytes_to_temp_file(root_certificates, '.crt')
+    _ca_file_cache_key = root_certificates
+    _ca_file_cache = path
+    return path
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    Custom CA prefers a process-lifetime temp snapshot of the validated PEM bytes
+    (never the resolved symlink target), so Kubernetes secret rotation that
+    removes the old ``..data`` version cannot invalidate an already-configured
+    session. If the snapshot cannot be written (e.g. read-only temp dir), fall
+    back to the still-readable configured CA path rather than discarding the
+    private CA for Requests' default trust store.
+
+    Client-cert temp-file failures drop mTLS only (one-way TLS), with a warning.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    if root_certificates is not None:
+        try:
+            verify: object = _ca_verify_temp_file(root_certificates)
+        except OSError as exc:
+            ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+            if ca_path is not None:
+                _warn_once(
+                    f'requests-ca-temp:{exc}',
+                    'Failed to persist trusted CA temp file (%s); '
+                    'falling back to configured CA path for HTTP verify.',
+                    exc,
+                )
+                verify = str(ca_path)
+            else:
+                _warn_once(
+                    f'requests-ca-temp:{exc}',
+                    'Failed to persist trusted CA temp file (%s); '
+                    'using process trust store for HTTP verify.',
+                    exc,
+                )
+                verify = True
+    else:
+        verify = True
+
+    if private_key is None or certificate_chain is None:
+        return verify, None
+
+    pair = _mtls_cert_key_files(certificate_chain, private_key)
+    return verify, pair
+
+
+def configure_requests_session(session) -> None:
+    try:
+        verify, cert = requests_tls_settings()
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'requests-tls:{exc}',
+            'Failed to configure HTTP reporter TLS (%s); falling back.',
+            exc,
+        )
+        if config.agent_force_tls or ssl_file_path(config.agent_ssl_trusted_ca_path):
+            # Prefer system-trust https over aborting agent start.
+            session.verify = True
+            session.cert = None
+        return
+    session.verify = verify
+    if cert is not None:
+        session.cert = cert
+
+
+def ssl_context_for_collector() -> Optional[ssl.SSLContext]:
+    """
+    stdlib SSLContext for aiohttp, or None when the collector stays plaintext.
+
+    Degrade on SSLError / OSError (never raise into agent bootstrap):
+    - bad / unparsable custom CA + FORCE_TLS → process trust store, no client cert
+    - bad / unparsable custom CA without FORCE_TLS → plaintext (None)
+    - bad client cert/key or temp-file failure → one-way TLS (CA or system trust)
+
+    Custom CA is loaded from normalized in-memory PEM bytes (``cadata``) when
+    present so path races and UTF-8 BOM/preamble cannot drop the custom trust
+    store after ``tls_pem_material`` already accepted the CA.
+    """
+    from skywalking import config
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+
+    def _system_trust_context() -> ssl.SSLContext:
+        return ssl.create_default_context()
+
+    ctx: Optional[ssl.SSLContext]
+    if root_certificates is not None:
+        load_exc: Optional[BaseException] = None
+        try:
+            # Normalized ASCII PEM from _load_trusted_ca (BOM/preamble stripped).
+            # Plain CERTIFICATE works via cadata; TRUSTED CERTIFICATE often needs cafile.
+            ctx = ssl.create_default_context(
+                cadata=root_certificates.decode('ascii'),
+            )
+        except (OSError, ValueError) as exc:

Review Comment:
   **[P2] Load mixed trusted-certificate bundles through cafile**
   
   For a bundle containing both `TRUSTED CERTIFICATE` and ordinary `CERTIFICATE` blocks, Python's `cadata` loader can successfully load the ordinary blocks while silently skipping the trusted ones. No exception is raised, so the new `cafile` fallback is never used. The asynchronous HTTP reporter then cannot verify a collector whose root is in a trusted-certificate block, even though direct `cafile` loading accepts the complete bundle. Skipping trusted blocks can also discard their trust restrictions.
   
   Please select the `cafile` loader whenever the bundle contains a trusted-certificate block, using the snapshot/configured-path fallback as appropriate. Success from `cadata` does not establish that every supported CA block was loaded.
   
   In an in-memory TLS handshake, direct `cafile` loading retained both entries and completed the handshake; `ssl_context_for_collector()` retained only the unrelated ordinary certificate and failed certificate verification.
   
   The following reproduction checks both the trust-store entries and the handshake. Run it from the PR checkout with project dependencies and OpenSSL installed (verified with Python 3.10). It uses existing test fixtures and requires no running server or open port. The fixture with an available private key serves as the in-memory TLS server identity; this checks client-side verification of the server, without OAP or mTLS client authentication:
   
   ```bash
   poetry run python - <<'PY'
   """Reproduce mixed CA loading with an in-memory TLS handshake."""
   import ssl
   import subprocess
   import tempfile
   from pathlib import Path
   
   from skywalking import config
   from skywalking.utils.tls import ssl_context_for_collector
   from tests.unit.test_tls import _TEST_CA_CERT, _TEST_CLIENT_CERT, _TEST_CLIENT_KEY_PKCS1
   
   
   def handshake(client_context, cert, key):
       server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
       server_context.load_cert_chain(str(cert), str(key))
       ci, co, si, so = (ssl.MemoryBIO() for _ in range(4))
       client = client_context.wrap_bio(ci, co, server_hostname='skywalking-test-client')
       server = server_context.wrap_bio(si, so, server_side=True)
       done = [False, False]
       for _ in range(20):
           for i, (peer, out, target) in enumerate(((client, co, si), (server, so, ci))):
               if not done[i]:
                   try:
                       peer.do_handshake()
                       done[i] = True
                   except ssl.SSLWantReadError:
                       pass
               data = out.read()
               if data:
                   target.write(data)
           if all(done):
               return 'OK'
       raise RuntimeError('handshake did not finish')
   
   
   with tempfile.TemporaryDirectory() as directory:
       root = Path(directory)
       ca, cert, key, trusted, bundle = (root / name for name in
                                        ('ca.pem', 'server.pem', 'server.key', 'trusted.pem', 'bundle.pem'))
       ca.write_bytes(_TEST_CA_CERT)
       cert.write_bytes(_TEST_CLIENT_CERT)
       key.write_bytes(_TEST_CLIENT_KEY_PKCS1)
       # Both fixtures are self-signed. Export the one with an available private key
       # as TRUSTED CERTIFICATE and use it as the local TLS server's identity.
       subprocess.run(['openssl', 'x509', '-in', str(cert), '-addtrust', 'serverAuth',
                       '-trustout', '-out', str(trusted)], check=True)
       bundle.write_bytes(trusted.read_bytes() + ca.read_bytes())
       config.agent_force_tls = False
       config.agent_ssl_trusted_ca_path = str(bundle)
       config.agent_ssl_cert_chain_path = ''
       config.agent_ssl_key_path = ''
       contexts = [('direct cafile', ssl.create_default_context(cafile=str(bundle))),
                   ('agent helper', ssl_context_for_collector())]
       for label, context in contexts:
           print(f'{label}: x509={context.cert_store_stats()["x509"]}')
           try:
               print(f'{label}: handshake={handshake(context, cert, key)}')
           except ssl.SSLCertVerificationError as exc:
               print(f'{label}: handshake=FAIL ({exc.verify_message})')
   PY
   ```
   
   Observed output:
   
   ```text
   direct cafile: x509=2
   direct cafile: handshake=OK
   agent helper: x509=1
   agent helper: handshake=FAIL (self-signed certificate)
   ```
   
   Expected: the agent context retains both certificates, matching direct `cafile` loading. The existing pure-`TRUSTED CERTIFICATE` case now works; please also cover this mixed-bundle case.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.

wu-sheng commented on code in PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#discussion_r3998190475


##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,738 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+
+    Returns ASCII PEM with only ``CERTIFICATE`` blocks extracted, so UTF-8 BOM
+    / preamble accepted by ``cafile`` remain usable for ``cadata`` and temp
+    verify files.
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return _extract_pem_blocks(data, 'CERTIFICATE')
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+
+    Only the bytes between the matching PKCS#1 BEGIN/END delimiters are
+    decoded — comments or ``openssl rsa -text`` preamble outside the block
+    must not corrupt the key (OpenSSL accepts such files).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    start = text.find(_PKCS1_PEM_HEADER)
+    end = text.find(_PKCS1_PEM_FOOTER, start)
+    if start < 0 or end < 0:
+        raise ValueError('Invalid PKCS#1 private key PEM: missing BEGIN/END delimiters')
+    body = text[start + len(_PKCS1_PEM_HEADER):end]
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body, validate=False)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _extract_pem_blocks(data: bytes, label: str) -> bytes:
+    """
+    Return ASCII PEM containing only ``BEGIN/END {label}`` blocks.
+
+    Strips UTF-8 BOM and ignores preamble/comments outside PEM delimiters so
+    OpenSSL-accepted files (BOM, UTF-8 comments) stay usable for ``cadata`` and
+    temp-file verify paths that require clean ASCII PEM.
+    """
+    text = data.decode('utf-8', errors='ignore').lstrip('\ufeff')
+    header = f'-----BEGIN {label}-----'
+    footer = f'-----END {label}-----'
+    blocks: List[str] = []
+    pos = 0
+    while True:
+        start = text.find(header, pos)
+        if start < 0:
+            break
+        end = text.find(footer, start)
+        if end < 0:
+            break
+        end += len(footer)
+        blocks.append(text[start:end].strip() + '\n')
+        pos = end
+    if not blocks:
+        raise ValueError(f'No {label} PEM block found')
+    return ''.join(blocks).encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS / CA temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_ca_file_cache_key: Optional[bytes] = None
+_ca_file_cache: Optional[str] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    global _ca_file_cache, _ca_file_cache_key, _mtls_file_cache, _mtls_file_cache_key
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    global _ca_file_cache, _ca_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def _ca_verify_temp_file(root_certificates: bytes) -> str:
+    """Process-lifetime temp path for trusted CA bytes (survives K8s secret rotation)."""
+    global _ca_file_cache_key, _ca_file_cache
+    if (
+        _ca_file_cache is not None
+        and _ca_file_cache_key == root_certificates
+        and os.path.exists(_ca_file_cache)
+    ):
+        return _ca_file_cache
+    path = _pem_bytes_to_temp_file(root_certificates, '.crt')
+    _ca_file_cache_key = root_certificates
+    _ca_file_cache = path
+    return path
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    Custom CA always uses a process-lifetime temp snapshot of the validated PEM
+    bytes (never the resolved symlink target), so Kubernetes secret rotation that
+    removes the old ``..data`` version cannot invalidate an already-configured
+    session.
+
+    Temp-file failures drop client certs only (one-way TLS), with a warning.
+    """
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    if root_certificates is not None:
+        try:
+            verify: object = _ca_verify_temp_file(root_certificates)
+        except OSError as exc:

Review Comment:
   **[P2] Preserve the readable custom CA when its snapshot cannot be written**
   
   The snapshot change introduces a writable-temp-directory requirement even for **one-way HTTP TLS**. If creating the snapshot raises `OSError`, this branch replaces the successfully loaded private CA with `verify=True` (Requests' default trust roots). A collector signed by that private CA then fails verification, despite the configured CA file still being readable. Before this update, the synchronous reporter used that readable file directly and needed no temporary writes. This affects, for example, a container with a read-only filesystem and no writable temporary mount.
   
   I reproduced the regression against a local HTTPS server using a genuinely non-writable directory, without mocking file creation: the previous helper returned the original CA path and the request succeeded with HTTP 200; the updated helper logged `PermissionError`, returned `verify=True`, and the request failed with `CERTIFICATE_VERIFY_FAILED`.
   
   Please preserve the loaded custom trust when snapshot creation fails, either through an in-memory context or a fallback to a still-readable configured CA path. The fallback should not discard a usable CA merely because a copy cannot be written.
   
   Minimal reproduction of the settings error, run from the PR checkout with dependencies installed, as a non-root user on Linux/macOS. It uses this PR's existing PEM fixture and needs no collector:
   
   ```bash
   poetry run python - <<'PY'
   import os
   import ssl
   import tempfile
   from pathlib import Path
   from skywalking import config
   from skywalking.utils import tls
   from tests.unit.test_tls import _TEST_CA_CERT
   
   config.agent_force_tls = False
   config.agent_ssl_cert_chain_path = ''
   config.agent_ssl_key_path = ''
   with tempfile.TemporaryDirectory() as tmp:
       ca = Path(tmp) / 'ca.crt'
       ca.write_bytes(_TEST_CA_CERT)
       config.agent_ssl_trusted_ca_path = str(ca)
       ssl.create_default_context(cafile=str(ca))
       blocked = Path(tmp) / 'readonly'
       blocked.mkdir()
       blocked.chmod(0o555)
       saved_tempdir = tempfile.tempdir
       try:
           if os.access(blocked, os.W_OK):
               raise SystemExit('Run as an unprivileged user on a POSIX filesystem.')
           tempfile.tempdir = str(blocked)
           print('Configured CA is readable:', ca.read_bytes() == _TEST_CA_CERT)
           print('Custom CA material loaded:', tls.tls_pem_material()[0] is not None)
           verify, cert = tls.requests_tls_settings()
           print('Requests verify:', verify)
           print('Client certificate:', cert)
       finally:
           tempfile.tempdir = saved_tempdir
           blocked.chmod(0o755)
   PY
   ```
   
   Observed output, plus the warning that the CA snapshot could not be written:
   
   ```text
   Configured CA is readable: True
   Custom CA material loaded: True
   Requests verify: True
   Client certificate: None
   ```
   
   Expected: Requests continues to use the readable private CA for one-way TLS. Running the equivalent setup on `a4dfb42` returns the CA file path instead of `True`.
   



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,738 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+
+    Returns ASCII PEM with only ``CERTIFICATE`` blocks extracted, so UTF-8 BOM
+    / preamble accepted by ``cafile`` remain usable for ``cadata`` and temp
+    verify files.
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return _extract_pem_blocks(data, 'CERTIFICATE')

Review Comment:
   **[P2] Preserve OpenSSL trusted-certificate blocks for synchronous HTTP**
   
   The new extraction only retains `CERTIFICATE` blocks, but OpenSSL also accepts `TRUSTED CERTIFICATE` PEMs produced by `openssl x509 -trustout`. A file containing only trusted-certificate blocks passes the `cafile` validation immediately above, then raises `No CERTIFICATE PEM block found` here. With `agent_force_tls=False`, the agent selects `http://`; with `agent_force_tls=True`, it discards the configured CA and falls back to default trust. Mixed bundles silently lose their trusted-certificate entries.
   
   This is a new regression for the **synchronous HTTP reporter**, which previously passed the OpenSSL-accepted CA file directly to Requests. The previous async `cadata` path already had a limitation with this format, so this finding is specifically about losing working synchronous HTTP support.
   
   Please retain supported trusted-certificate blocks **and their trust attributes** in the Requests CA snapshot. Merely relabeling them as `CERTIFICATE` can discard trust restrictions; normalization should not treat valid trust-store entries as preamble.
   
   Reproduction from the PR checkout, with project dependencies and OpenSSL installed:
   
   ```bash
   poetry run python - <<'PY'
   import ssl
   import subprocess
   import tempfile
   from pathlib import Path
   from skywalking import config
   from skywalking.utils import tls
   from tests.unit.test_tls import _TEST_CA_CERT
   
   config.agent_force_tls = False
   config.agent_ssl_cert_chain_path = ''
   config.agent_ssl_key_path = ''
   with tempfile.TemporaryDirectory() as tmp:
       ca = Path(tmp) / 'ca.pem'
       trusted = Path(tmp) / 'trusted.pem'
       ca.write_bytes(_TEST_CA_CERT)
       subprocess.run([
           'openssl', 'x509', '-in', str(ca), '-addtrust', 'serverAuth',
           '-trustout', '-out', str(trusted),
       ], check=True)
       ssl.create_default_context(cafile=str(trusted))
       print('OpenSSL cafile accepted:', trusted.read_text().splitlines()[0])
       config.agent_ssl_trusted_ca_path = str(trusted)
       print('Agent TLS material:', tls.tls_pem_material())
       print('Agent HTTP scheme:', tls.collector_http_scheme())
   PY
   ```
   
   Observed output, plus the warning about `No CERTIFICATE PEM block found`:
   
   ```text
   OpenSSL cafile accepted: -----BEGIN TRUSTED CERTIFICATE-----
   Agent TLS material: None
   Agent HTTP scheme: http://
   ```
   
   Expected: the configured, accepted trusted CA remains usable for synchronous HTTPS. On `a4dfb42`, the same configuration returns TLS material, selects `https://`, and uses the trusted CA file as Requests' verify path.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat: gRPC/HTTP collector TLS and mTLS [skywalking-python]

Posted by "songzhendong (via GitHub)" <gi...@apache.org>.

songzhendong commented on code in PR #419:
URL: https://github.com/apache/skywalking-python/pull/419#discussion_r3998560614


##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,738 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+
+    Returns ASCII PEM with only ``CERTIFICATE`` blocks extracted, so UTF-8 BOM
+    / preamble accepted by ``cafile`` remain usable for ``cadata`` and temp
+    verify files.
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return _extract_pem_blocks(data, 'CERTIFICATE')
+
+
+def _validate_client_cert_key(cert_pem: bytes, key_pem: bytes) -> None:
+    """Raise OSError / ValueError when cert+key cannot load as a client identity."""
+    cert_path = key_path = None
+    try:
+        fd, cert_path = tempfile.mkstemp(suffix='.crt')
+        try:
+            os.write(fd, cert_pem)
+        finally:
+            os.close(fd)
+        fd, key_path = tempfile.mkstemp(suffix='.pem')
+        try:
+            os.write(fd, key_pem)
+        finally:
+            os.close(fd)
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        ctx.load_cert_chain(cert_path, key_path)
+    finally:
+        for path in (cert_path, key_path):
+            if path is None:
+                continue
+            try:
+                os.unlink(path)
+            except OSError:
+                pass
+
+
+# OpenSSL PKCS#1 PEM → PKCS#8 PEM for stacks that only accept PKCS#8.
+# grpcio/BoringSSL and CPython ssl both accept PKCS#1 RSA PEMs directly; this
+# wrap exists for HTTP stacks / older OpenSSL bindings that reject
+# ``BEGIN RSA PRIVATE KEY`` and only load PKCS#8 ``BEGIN PRIVATE KEY``.
+_PKCS1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+_PKCS1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----'
+_PKCS8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----'
+_PKCS8_PEM_FOOTER = '-----END PRIVATE KEY-----'
+_ENCRYPTED_PEM_HEADER = '-----BEGIN ENCRYPTED PRIVATE KEY-----'
+
+
+def normalize_private_key_pem(key_pem: bytes) -> bytes:
+    """
+    Return PKCS#8 PEM bytes. PKCS#1 (`BEGIN RSA PRIVATE KEY`) is wrapped into
+    an unencrypted PKCS#8 PEM; other PEM/DER forms pass through.
+
+    Passphrase-encrypted PEMs (PKCS#8 encrypted or legacy OpenSSL Proc-Type)
+    are rejected with a clear error (not supported).
+
+    Only the bytes between the matching PKCS#1 BEGIN/END delimiters are
+    decoded — comments or ``openssl rsa -text`` preamble outside the block
+    must not corrupt the key (OpenSSL accepts such files).
+    """
+    text = key_pem.decode('utf-8', errors='ignore')
+    if _ENCRYPTED_PEM_HEADER in text:
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    # Legacy OpenSSL ``openssl genrsa -aes256`` / ``rsa -des3`` output.
+    if 'Proc-Type:' in text and 'ENCRYPTED' in text.upper():
+        raise ValueError(
+            'Passphrase-encrypted private keys are not supported; '
+            'use an unencrypted PEM (PKCS#8 or PKCS#1 RSA)'
+        )
+    if _PKCS1_PEM_HEADER not in text:
+        return key_pem
+
+    start = text.find(_PKCS1_PEM_HEADER)
+    end = text.find(_PKCS1_PEM_FOOTER, start)
+    if start < 0 or end < 0:
+        raise ValueError('Invalid PKCS#1 private key PEM: missing BEGIN/END delimiters')
+    body = text[start + len(_PKCS1_PEM_HEADER):end]
+    body = body.replace('\r', '').replace('\n', '').replace(' ', '')
+    try:
+        pkcs1 = base64.b64decode(body, validate=False)
+    except ValueError as exc:
+        raise ValueError(f'Invalid PKCS#1 private key PEM: {exc}') from exc
+
+    pkcs1_length = len(pkcs1)
+    total_length = pkcs1_length + 22
+    # Fixed 2-byte length form (keys up to ~64KiB DER).
+    if total_length > 0xFFFF or pkcs1_length > 0xFFFF:
+        raise ValueError('PKCS#1 private key too large for PKCS#8 wrap')
+
+    # Hand-rolled PKCS#8 PrivateKeyInfo wrapping RSA PKCS#1 (AlgorithmIdentifier
+    # rsaEncryption). Verified byte-identical to cryptography for 2048/4096 RSA.
+    pkcs8_header = bytes([
+        0x30, 0x82, (total_length >> 8) & 0xFF, total_length & 0xFF,
+        0x02, 0x01, 0x00,
+        0x30, 0x0D, 0x06, 0x09, 0x2A, 0x86, 0x48, 0x86, 0xF7, 0x0D, 0x01, 0x01, 0x01, 0x05, 0x00,
+        0x04, 0x82, (pkcs1_length >> 8) & 0xFF, pkcs1_length & 0xFF,
+    ])
+    encoded = base64.b64encode(pkcs8_header + pkcs1).decode('ascii')
+    # 64-char lines, matching typical PEM layout.
+    lines = [encoded[i:i + 64] for i in range(0, len(encoded), 64)]
+    pem = _PKCS8_PEM_HEADER + '\n' + '\n'.join(lines) + '\n' + _PKCS8_PEM_FOOTER + '\n'
+    return pem.encode('ascii')
+
+
+def _extract_pem_blocks(data: bytes, label: str) -> bytes:
+    """
+    Return ASCII PEM containing only ``BEGIN/END {label}`` blocks.
+
+    Strips UTF-8 BOM and ignores preamble/comments outside PEM delimiters so
+    OpenSSL-accepted files (BOM, UTF-8 comments) stay usable for ``cadata`` and
+    temp-file verify paths that require clean ASCII PEM.
+    """
+    text = data.decode('utf-8', errors='ignore').lstrip('\ufeff')
+    header = f'-----BEGIN {label}-----'
+    footer = f'-----END {label}-----'
+    blocks: List[str] = []
+    pos = 0
+    while True:
+        start = text.find(header, pos)
+        if start < 0:
+            break
+        end = text.find(footer, start)
+        if end < 0:
+            break
+        end += len(footer)
+        blocks.append(text[start:end].strip() + '\n')
+        pos = end
+    if not blocks:
+        raise ValueError(f'No {label} PEM block found')
+    return ''.join(blocks).encode('ascii')
+
+
+def _mtls_material(*, ca_usable: bool) -> Tuple[Optional[bytes], Optional[bytes]]:
+    """
+    Client certificate_chain and private_key PEM bytes, or (None, None).
+
+    mTLS is considered only when a custom CA was successfully loaded and parsed
+    (``ca_usable``). A CA path that exists but failed to load must not unlock
+    client certs for FORCE_TLS / system-trust fallback.
+    Private keys are normalized to PKCS#8 when the file is PKCS#1.
+    """
+    from skywalking import config
+
+    cert_cfg = _configured_path(config.agent_ssl_cert_chain_path)
+    key_cfg = _configured_path(config.agent_ssl_key_path)
+
+    if not ca_usable:
+        # Client material without a usable CA cannot enable mTLS (by design).
+        if cert_cfg or key_cfg:
+            _warn_once(
+                f'mtls-without-ca:{cert_cfg}|{key_cfg}',
+                'Client SSL cert/key is configured but SW_AGENT_SSL_TRUSTED_CA_PATH '
+                'is missing, unreadable, or not a valid CA PEM; mTLS is disabled. '
+                'Provide a readable CA PEM to enable mTLS (FORCE_TLS alone uses '
+                'system trust for one-way TLS).',
+            )
+        return None, None
+
+    if bool(cert_cfg) ^ bool(key_cfg):
+        _warn_once(
+            f'mtls-partial:{cert_cfg}|{key_cfg}',
+            'Only one of SW_AGENT_SSL_CERT_CHAIN_PATH / SW_AGENT_SSL_KEY_PATH is set; '
+            'mTLS requires both. Staying on one-way TLS.',
+        )
+        return None, None
+
+    if not cert_cfg or not key_cfg:
+        return None, None
+
+    cert_path = ssl_file_path(cert_cfg)
+    key_path = ssl_file_path(key_cfg)
+    if cert_path is None or key_path is None:
+        logger.warning('Failed to enable mTLS caused by cert or key cannot be found.')
+        return None, None
+
+    try:
+        cert_pem = _read_bytes(cert_path)
+        key_pem = normalize_private_key_pem(_read_bytes(key_path))
+        _validate_client_cert_key(cert_pem, key_pem)
+        return cert_pem, key_pem
+    except (OSError, ValueError) as exc:
+        logger.warning('Failed to enable mTLS caused by cert or key read error: %s', exc)
+        return None, None
+
+
+# Keep mTLS / CA temp PEM paths alive for the process (requests/ssl need file paths).
+_mtls_temp_files: List[str] = []
+_mtls_file_cache_key: Optional[Tuple[bytes, bytes]] = None
+_mtls_file_cache: Optional[Tuple[str, str]] = None
+_ca_file_cache_key: Optional[bytes] = None
+_ca_file_cache: Optional[str] = None
+_atexit_registered = False
+
+
+def _cleanup_mtls_temp_files() -> None:
+    global _ca_file_cache, _ca_file_cache_key, _mtls_file_cache, _mtls_file_cache_key
+    for path in list(_mtls_temp_files):
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+    _mtls_temp_files.clear()
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+def _after_fork_in_child() -> None:
+    """
+    Drop inherited temp-path bookkeeping without unlinking.
+
+    Parent-created PEM temps must stay on disk for the parent reporter. Clearing
+    the list in-place would also empty the parent's list (same object); rebind
+    instead so the child's atexit handler cannot delete the parent's files.
+    """
+    global _mtls_temp_files, _mtls_file_cache, _mtls_file_cache_key
+    global _ca_file_cache, _ca_file_cache_key
+    _mtls_temp_files = []
+    _mtls_file_cache = None
+    _mtls_file_cache_key = None
+    _ca_file_cache = None
+    _ca_file_cache_key = None
+
+
+if hasattr(os, 'register_at_fork'):
+    os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
+def _ensure_atexit() -> None:
+    global _atexit_registered
+    if not _atexit_registered:
+        atexit.register(_cleanup_mtls_temp_files)
+        _atexit_registered = True
+
+
+def _pem_bytes_to_temp_file(data: bytes, suffix: str) -> str:
+    _ensure_atexit()
+    handle, path = tempfile.mkstemp(suffix=suffix)
+    try:
+        view = memoryview(data)
+        offset = 0
+        while offset < len(view):
+            written = os.write(handle, view[offset:])
+            if written <= 0:
+                raise OSError(f'Failed to write SSL PEM temp file {path}')
+            offset += written
+    except Exception:
+        os.close(handle)
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+        raise
+    else:
+        os.close(handle)
+    _mtls_temp_files.append(path)
+    return path
+
+
+def _mtls_cert_key_files(
+    certificate_chain: bytes,
+    private_key: bytes,
+) -> Optional[Tuple[str, str]]:
+    """
+    Write normalized PEM bytes to temp files (cached per material).
+
+    Returns None when temp files cannot be created (caller stays one-way TLS).
+    """
+    global _mtls_file_cache_key, _mtls_file_cache
+
+    cache_key = (certificate_chain, private_key)
+    if _mtls_file_cache is not None and _mtls_file_cache_key == cache_key:
+        return _mtls_file_cache
+
+    try:
+        paths = (
+            _pem_bytes_to_temp_file(certificate_chain, '.crt'),
+            _pem_bytes_to_temp_file(private_key, '.pem'),
+        )
+    except OSError as exc:
+        _warn_once(
+            f'mtls-temp:{exc}',
+            'Failed to write mTLS cert/key temp files (%s); staying on one-way TLS.',
+            exc,
+        )
+        return None
+
+    _mtls_file_cache_key = cache_key
+    _mtls_file_cache = paths
+    return paths
+
+
+def tls_pem_material() -> Optional[Tuple[Optional[bytes], Optional[bytes], Optional[bytes]]]:
+    """
+    PEM bytes for (root_certificates, private_key, certificate_chain).
+
+    None means plaintext (no TLS). Tuple members may still be None when using
+    FORCE_TLS with the process trust store and/or one-way TLS.
+
+    CA / client PEMs are parsed before return so invalid content degrades here
+    (plaintext or system-trust one-way TLS) instead of only failing at connect.
+
+    Enable decision uses the *current* CA load result (not a prior existence
+    check alone): if the CA path races away between resolve and read without
+    FORCE_TLS, returns None (plaintext) rather than a system-trust HTTPS tuple.
+    """
+    from skywalking import config
+
+    # Emit configured-but-unusable CA path warnings (warn_once dedupes).
+    collector_uses_tls()
+
+    force = bool(config.agent_force_tls)
+    ca_path = ssl_file_path(config.agent_ssl_trusted_ca_path)
+    root_certificates = None
+    ca_usable = False
+    if ca_path is not None:
+        try:
+            root_certificates = _load_trusted_ca(ca_path)
+            ca_usable = True
+        except (OSError, ValueError) as exc:
+            if force:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); continuing with FORCE_TLS '
+                    'and the process trust store (custom CA ignored, mTLS disabled).',
+                    exc,
+                )
+            else:
+                logger.warning(
+                    'Failed to load trusted CA file (%s); collector stays plaintext '
+                    '(set SW_AGENT_FORCE_TLS to use the process trust store instead).',
+                    exc,
+                )
+                # Still surface mTLS-without-CA diagnostics when client PEMs are set.
+                _mtls_material(ca_usable=False)
+                return None
+
+    if not force and not ca_usable:
+        # No TLS without FORCE or a usable CA — including CA that vanished after
+        # collector_uses_tls() saw it (TOCTOU). Warn if client cert/key are set.
+        _mtls_material(ca_usable=False)
+        return None
+
+    certificate_chain, private_key = _mtls_material(ca_usable=ca_usable)
+    return root_certificates, private_key, certificate_chain
+
+
+def grpc_ssl_credentials():
+    """
+    grpc.ChannelCredentials for TLS/mTLS, or None for plaintext.
+
+    FORCE_TLS without a CA file uses the process trust store
+    (grpc.ssl_channel_credentials() with no roots).
+
+    If credential construction ultimately fails (including process-trust
+    FORCE_TLS fallback), returns None so the channel stays plaintext with a
+    warning — never raises into agent start.
+    """
+    import grpc
+
+    material = tls_pem_material()
+    if material is None:
+        return None
+
+    root_certificates, private_key, certificate_chain = material
+    try:
+        return grpc.ssl_channel_credentials(
+            root_certificates=root_certificates,
+            private_key=private_key,
+            certificate_chain=certificate_chain,
+        )
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        if private_key is not None or certificate_chain is not None:
+            _warn_once(
+                f'grpc-creds-mtls:{exc}',
+                'Failed to build gRPC mTLS credentials (%s); retrying one-way TLS.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials(root_certificates=root_certificates)
+            except Exception as one_way_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-oneway:{one_way_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    one_way_exc,
+                )
+                return None
+        if config.agent_force_tls:
+            _warn_once(
+                f'grpc-creds-force:{exc}',
+                'Failed to build gRPC TLS credentials with custom CA (%s); '
+                'continuing with FORCE_TLS and the process trust store.',
+                exc,
+            )
+            try:
+                return grpc.ssl_channel_credentials()
+            except Exception as force_exc:  # noqa: BLE001
+                _warn_once(
+                    f'grpc-creds-force-fail:{force_exc}',
+                    'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+                    force_exc,
+                )
+                return None
+        _warn_once(
+            f'grpc-creds:{exc}',
+            'Failed to build gRPC TLS credentials (%s); collector stays plaintext.',
+            exc,
+        )
+        return None
+
+
+def collector_http_scheme() -> str:
+    """
+    ``https://`` when TLS material is enabled, else ``http://``.
+
+    Never raises into reporter ``__init__`` (HTTP clients call this before
+    other TLS helpers that may wrap failures).
+    """
+    try:
+        return 'https://' if tls_pem_material() is not None else 'http://'
+    except Exception as exc:  # noqa: BLE001 - never fail host process start
+        from skywalking import config
+
+        _warn_once(
+            f'http-scheme:{exc}',
+            'Failed to decide collector HTTP scheme (%s); falling back.',
+            exc,
+        )
+        return 'https://' if config.agent_force_tls else 'http://'
+
+
+def _ca_verify_temp_file(root_certificates: bytes) -> str:
+    """Process-lifetime temp path for trusted CA bytes (survives K8s secret rotation)."""
+    global _ca_file_cache_key, _ca_file_cache
+    if (
+        _ca_file_cache is not None
+        and _ca_file_cache_key == root_certificates
+        and os.path.exists(_ca_file_cache)
+    ):
+        return _ca_file_cache
+    path = _pem_bytes_to_temp_file(root_certificates, '.crt')
+    _ca_file_cache_key = root_certificates
+    _ca_file_cache = path
+    return path
+
+
+def requests_tls_settings() -> Tuple[object, Optional[Tuple[str, str]]]:
+    """
+    (verify, cert) for requests.Session.
+
+    verify is True (system CAs), a CA file path, or unused for plaintext callers.
+    cert is (cert_path, key_path) when mTLS files are present.
+
+    Keeps the same enable/disable decision as grpc_ssl_credentials / tls_pem_material
+    so an unreadable or oversized CA cannot leave HTTP on https:// with a bad verify path.
+
+    Custom CA always uses a process-lifetime temp snapshot of the validated PEM
+    bytes (never the resolved symlink target), so Kubernetes secret rotation that
+    removes the old ``..data`` version cannot invalidate an already-configured
+    session.
+
+    Temp-file failures drop client certs only (one-way TLS), with a warning.
+    """
+    material = tls_pem_material()
+    if material is None:
+        return True, None
+
+    root_certificates, private_key, certificate_chain = material
+    if root_certificates is not None:
+        try:
+            verify: object = _ca_verify_temp_file(root_certificates)
+        except OSError as exc:

Review Comment:
   Fixed in `1327f8b`: fall back to the configured CA path when the snapshot cannot be written; covered by `test_requests_ca_temp_failure_falls_back_to_configured_path`.



##########
skywalking/utils/tls.py:
##########
@@ -0,0 +1,738 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""
+Collector TLS / mTLS helpers for gRPC and HTTP reporters.
+
+Rules:
+- TLS when FORCE_TLS is set or the trusted CA file exists.
+- Custom CA is used only when that file exists *and* parses as a CA PEM
+  (otherwise system trust with FORCE_TLS, or plaintext without it).
+- mTLS (client cert + key) is enabled only when a usable custom CA was loaded
+  *and* both cert-chain and key files exist and parse. Missing / invalid
+  cert/key logs a warning and stays one-way TLS — it does not fail agent start.
+- A non-empty CA / cert / key path that cannot be resolved logs a warning
+  (misconfig must not look like "TLS off by design").
+- Invalid PEM content, SSLContext build failures, and temp-file I/O errors
+  never raise into the host process: warn and degrade (see degrade table in
+  ``ssl_context_for_collector`` / ``requests_tls_settings``).
+- HTTP mTLS temp PEMs are process-local: after ``os.fork()`` the child rebinds
+  its temp-file list so exiting the child cannot unlink the parent's files.
+"""
+
+from __future__ import annotations
+
+import atexit
+import base64
+import os
+import ssl
+import tempfile
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+from skywalking.loggings import logger
+
+# Align with Node TlsMaterialCache: refuse oversized PEMs (DoS / misconfig).
+_MAX_PEM_BYTES = 256 * 1024
+# Avoid repeating the same misconfig warning on every reporter call.
+_warned_keys: set = set()
+
+
+def _warn_once(key: str, message: str, *args) -> None:
+    if key in _warned_keys:
+        return
+    _warned_keys.add(key)
+    logger.warning(message, *args)
+
+
+def _configured_path(value: str) -> str:
+    return (value or '').strip()
+
+
+def ssl_file_path(value: str) -> Optional[Path]:
+    """
+    Return a readable regular file path, or None.
+
+    Paths are cwd-relative or absolute. Symlinks are followed (Kubernetes
+    secret mounts are typically ``ca.crt -> ..data/ca.crt``); the resolved
+    target must be a regular file.
+
+    Never raises: ``expanduser`` / ``resolve`` failures (including symlink
+    loops and unresolvable ``~user`` → ``RuntimeError``) degrade to None so
+    TLS misconfig cannot abort agent start.
+    """
+    text = _configured_path(value)
+    if not text:
+        return None
+    try:
+        path = Path(text).expanduser()
+        resolved = path.resolve(strict=True)
+        if resolved.is_file():
+            return resolved
+    except (OSError, ValueError, RuntimeError):
+        # OSError: missing path / I/O. ValueError: embedded NUL. RuntimeError:
+        # expanduser ~user miss or symlink-loop resolve on some platforms.
+        return None
+    return None
+
+
+def collector_uses_tls() -> bool:
+    from skywalking import config
+
+    force = bool(config.agent_force_tls)
+    ca_cfg = _configured_path(config.agent_ssl_trusted_ca_path)
+    ca_path = ssl_file_path(ca_cfg) if ca_cfg else None
+    if ca_cfg and ca_path is None:
+        if force:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; custom CA is ignored. FORCE_TLS remains on with the process trust store.',
+                ca_cfg,
+            )
+        else:
+            _warn_once(
+                f'ca-missing:{ca_cfg}',
+                'SW_AGENT_SSL_TRUSTED_CA_PATH is set (%r) but is not a readable regular '
+                'file; TLS via custom CA is disabled and the collector stays plaintext '
+                '(set SW_AGENT_FORCE_TLS to use the process trust store).',
+                ca_cfg,
+            )
+    return force or ca_path is not None
+
+
+def _read_bytes(path: Path) -> bytes:
+    size = os.path.getsize(path)
+    if size > _MAX_PEM_BYTES:
+        raise ValueError(
+            f'SSL PEM file exceeds {_MAX_PEM_BYTES} bytes: {path} ({size} bytes)'
+        )
+    return path.read_bytes()
+
+
+def _load_trusted_ca(path: Path) -> bytes:
+    """
+    Read and parse a trusted CA PEM.
+
+    Raises OSError / ValueError when the file is unreadable, oversized, or not
+    a CA bundle OpenSSL can load — callers degrade instead of handing garbage
+    to gRPC / requests (which may only fail at connect time).
+
+    Returns ASCII PEM with only ``CERTIFICATE`` blocks extracted, so UTF-8 BOM
+    / preamble accepted by ``cafile`` remain usable for ``cadata`` and temp
+    verify files.
+    """
+    data = _read_bytes(path)
+    try:
+        ssl.create_default_context(cafile=str(path))
+    except OSError as exc:
+        # ssl.SSLError subclasses OSError on CPython.
+        raise ValueError(f'Invalid trusted CA PEM {path}: {exc}') from exc
+    return _extract_pem_blocks(data, 'CERTIFICATE')

Review Comment:
   Fixed in `1327f8b`: retain `TRUSTED CERTIFICATE` blocks and trust attributes in the CA snapshot; covered by `test_trusted_certificate_pem_enables_sync_http_tls`.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org