Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/docker-hub-publish.yml
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,15 @@ jobs:

- name: Build the Docker image

run: docker build . --file Dockerfile -t mobilesecurity/mdast_cli:2026.7.2 -t mobilesecurity/mdast_cli:latest
run: docker build . --file Dockerfile -t mobilesecurity/mdast_cli:2026.8.1 -t mobilesecurity/mdast_cli:latest


- name: Docker Hub push latest image
run: docker push mobilesecurity/mdast_cli:latest

- name: Docker Hub push tagged image

run: docker push mobilesecurity/mdast_cli:2026.7.2
run: docker push mobilesecurity/mdast_cli:2026.8.1



Expand Down
33 changes: 33 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -219,6 +219,39 @@ preferable to a probe:
export MDAST_CLI_MODE=microservices # or: monolith (default: auto)
```

> **Note for SaaS users:** this setting describes the **stand you connect to**, not
> anything deployed in your own infrastructure. If you use the hosted service, you
> do not need to install anything - just set the value matching your stand (ask us
> if unsure).

### Behind a TLS-inspecting proxy

Corporate proxies that inspect TLS re-sign traffic with an internal CA, which Python
does not trust by default. The symptom is:

```
certificate verify failed: self-signed certificate in certificate chain
```

Point `REQUESTS_CA_BUNDLE` at your corporate root CA (PEM format):

```bash
export REQUESTS_CA_BUNDLE=/path/to/corporate-ca.pem
```

**Adding the CA to the OS trust store is not enough.** The CLI uses `requests`,
which reads the `certifi` bundle rather than the system store - `update-ca-certificates`
alone will not help. To keep trusting public CAs as well, concatenate the two:

```bash
cat $(python3 -c "import certifi; print(certifi.where())") corporate-ca.pem > ca-bundle.pem
export REQUESTS_CA_BUNDLE=ca-bundle.pem
```

If the CA is not available, the CLI still proceeds: mode detection retries without
certificate verification and logs a warning. Setting `MDAST_CLI_MODE` explicitly skips
the probe altogether.

### Environment variables (microservices installation)

| Variable | Default | Purpose |
Expand Down
2 changes: 1 addition & 1 deletion mdast_cli/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +1 @@
__version__ = '2026.7.1'
__version__ = '2026.8.1'
149 changes: 132 additions & 17 deletions mdast_cli_core/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
"""
import logging
import os
from collections import namedtuple

import requests

Expand All @@ -28,6 +29,14 @@
MODE_MICROSERVICES = 'microservices'
PROBE_TIMEOUT = 20

# Why a probe did not complete. Distinguished so the error message can point at
# the actual cause instead of blaming --url/--token for every transport failure.
FAILURE_TLS = 'tls'
FAILURE_PROXY = 'proxy'
FAILURE_NETWORK = 'network'

Probe = namedtuple('Probe', 'status payload failure detail')

logger = logging.getLogger(__name__)


Expand Down Expand Up @@ -78,22 +87,106 @@ def _looks_monolith(payload):


def _probe(base_url, path, scheme, ci_token, verify):
"""GET a probe endpoint; return (status_code | None, parsed_json | None)."""
"""GET a probe endpoint.

Returns a :class:`Probe`. ``failure`` is ``None`` when the request completed
(whatever the status code) and one of the ``FAILURE_*`` constants when it did
not, so the caller can tell a TLS problem from a dead proxy instead of
reporting every transport error as a bad --url/--token.
"""
try:
resp = requests.get(f'{base_url}{path}',
headers={'Authorization': f'{scheme} {ci_token}'},
verify=verify,
timeout=PROBE_TIMEOUT)
except requests.exceptions.SSLError as ex:
logger.debug(f'Probe {scheme} {path} failed: SSLError: {ex}')
return Probe(None, None, FAILURE_TLS, str(ex))
except requests.exceptions.ProxyError as ex:
logger.debug(f'Probe {scheme} {path} failed: ProxyError: {ex}')
return Probe(None, None, FAILURE_PROXY, str(ex))
except requests.RequestException as ex:
logger.debug(f'Probe {scheme} {path} failed: {type(ex).__name__}: {ex}')
return None, None
return Probe(None, None, FAILURE_NETWORK, f'{type(ex).__name__}: {ex}')
payload = None
detail = None
if resp.status_code == 200:
try:
payload = resp.json()
except ValueError:
payload = None
return resp.status_code, payload
elif resp.status_code in (401, 403):
# The server explains itself ("Token has expired", ...) - keep the text so
# the operator does not have to guess which half of the auth pair is wrong.
detail = _body_snippet(resp, ci_token)
logger.debug(f'Probe {scheme} {path} -> {resp.status_code}, body: {detail}')
return Probe(resp.status_code, payload, None, detail)


def _body_snippet(resp, ci_token=None, limit=300):
"""Short single-line excerpt of a response body, for error messages.

The token is redacted: this text is surfaced in exceptions and logs, and a
server that echoes the credential back must not turn that into a leak.
"""
try:
text = resp.text or ''
except Exception: # pragma: no cover - defensive, .text should not raise
return None
text = ' '.join(text.split())
if ci_token:
text = text.replace(ci_token, '***')
if not text:
return None
return text[:limit] + ('...' if len(text) > limit else '')


def _run_probes(base_url, ci_token, verify):
"""Probe both flavours once; return ``(mode | None, ms_probe, mono_probe)``."""
# Microservices probe: Bearer + architectures, classify by payload shape.
ms = _probe(base_url, '/architectures/', 'Bearer', ci_token, verify)
if ms.status == 200 and _looks_microservices(ms.payload):
return MODE_MICROSERVICES, ms, None

# Monolith probe: Token + architectures, classify by payload shape.
mono = _probe(base_url, '/architectures/', 'Token', ci_token, verify)
if mono.status == 200 and _looks_monolith(mono.payload):
return MODE_MONOLITH, ms, mono

return None, ms, mono


def _failures(*probes):
"""Set of failure kinds seen across the given probes."""
return {p.failure for p in probes if p is not None and p.failure}


def _diagnosis(ms, mono):
"""Actionable explanation for a failed detection, tailored to the cause.

Every transport error used to surface as 'Check --url/--token', which sends
the operator after a credential that is usually fine. Name the real cause.
"""
failures = _failures(ms, mono)
if FAILURE_TLS in failures:
return ('The certificate could not be verified even with verification disabled, '
'so the TLS handshake itself is failing. Check the proxy/TLS setup, or '
f'force the mode via {MODE_ENV_VAR}=monolith|microservices.')
if FAILURE_PROXY in failures:
return ('The proxy is unreachable. Check HTTP_PROXY/HTTPS_PROXY/ALL_PROXY - they '
'are often left set after a VPN is switched off; unset them to connect '
'directly.')
if FAILURE_NETWORK in failures:
detail = (ms.detail if ms is not None and ms.failure else None) or \
(mono.detail if mono is not None and mono.failure else None)
suffix = f' ({detail})' if detail else ''
return f'The host could not be reached{suffix}. Check --url and network access.'

# Both probes completed: this is a genuine HTTP-level answer, so quote it.
details = [p.detail for p in (ms, mono) if p is not None and p.detail]
if details:
return f'Server said: {details[0]} - check --token (and --url).'
return f'Check --url/--token, or force the mode via {MODE_ENV_VAR}.'


def resolve_installation_mode(base_url, ci_token, company_id, mode=None, verify=None):
Expand All @@ -113,21 +206,43 @@ def resolve_installation_mode(base_url, ci_token, company_id, mode=None, verify=
f'Unknown {MODE_ENV_VAR} value: {mode!r} '
f'(expected {MODE_AUTO}/{MODE_MONOLITH}/{MODE_MICROSERVICES})')

# Microservices probe: Bearer + architectures, classify by payload shape.
ms_status, ms_payload = _probe(base_url, '/architectures/', 'Bearer', ci_token, verify)
if ms_status == 200 and _looks_microservices(ms_payload):
detected, ms, mono = _run_probes(base_url, ci_token, verify)
if detected == MODE_MICROSERVICES:
logger.info('Detected microservices installation (Clark facade)')
return MODE_MICROSERVICES

# Monolith probe: Token + architectures, classify by payload shape.
mono_status, mono_payload = _probe(base_url, '/architectures/', 'Token', ci_token, verify)
if mono_status == 200 and _looks_monolith(mono_payload):
return detected
if detected == MODE_MONOLITH:
logger.info('Detected monolith installation')
return MODE_MONOLITH
return detected

# TLS fallback: the monolith flow itself does not verify certificates, so a
# certificate that only this probe rejects must not block the whole run -
# that would fail at the door of a room with no walls (typical cause: a
# corporate TLS-inspecting proxy re-signing with an internal CA).
if verify and FAILURE_TLS in _failures(ms, mono):
logger.warning(
'TLS certificate verification failed while detecting the installation mode. '
'Retrying the detection probe WITHOUT certificate verification. '
f'This usually means a TLS-inspecting proxy is in the path - point '
f'REQUESTS_CA_BUNDLE at your corporate root CA (in PEM; note that requests '
f'uses the certifi bundle, NOT the OS trust store) to silence this.')
insecure, ms_insecure, mono_insecure = _run_probes(base_url, ci_token, False)
if insecure == MODE_MONOLITH:
logger.info('Detected monolith installation (certificate not verified)')
return MODE_MONOLITH
if insecure == MODE_MICROSERVICES:
# A microservices install DOES verify certificates for real traffic, so
# resolving its mode over an unverified channel buys nothing - the scan
# would fail moments later. Ask for an explicit decision instead.
raise ModeDetectionError(
'Detected a microservices installation, but only over a connection whose '
'certificate could not be verified. Point REQUESTS_CA_BUNDLE at your '
f'corporate root CA, or set {TLS_VERIFY_ENV_VAR}=0 to accept the risk '
f'(the organization CI token would be sent over an unverified connection).')
ms, mono = ms_insecure, mono_insecure

# Ambiguous 200 (payload matched neither shape) — do not guess.
if ms_status == 200 or mono_status == 200:
empty_list = ms_payload == [] or mono_payload == []
if ms.status == 200 or mono.status == 200:
empty_list = ms.payload == [] or mono.payload == []
hint = ('The /architectures/ list is empty, so the installation flavour cannot be '
'inferred from it. ' if empty_list else
'The payload matched neither the microservices (string type + os_version) '
Expand All @@ -136,9 +251,9 @@ def resolve_installation_mode(base_url, ci_token, company_id, mode=None, verify=
f'Cannot detect installation mode: /architectures/ returned 200 but {hint}'
f'Force the mode via {MODE_ENV_VAR}=monolith|microservices.')

auth_error = 401 in (ms_status, mono_status) or 403 in (ms_status, mono_status)
auth_error = 401 in (ms.status, mono.status) or 403 in (ms.status, mono.status)
raise ModeDetectionError(
'Cannot detect installation mode via GET /architectures/ '
f'(Bearer -> {ms_status}, Token -> {mono_status}). '
f'Check --url/--token, or force the mode via {MODE_ENV_VAR}.',
f'(Bearer -> {ms.status}, Token -> {mono.status}). '
f'{_diagnosis(ms, mono)}',
auth_error=auth_error)
2 changes: 1 addition & 1 deletion setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
setup(
name="mdast_cli",

version='2026.7.2',
version='2026.8.1',

python_requires='>=3.12',

Expand Down
83 changes: 83 additions & 0 deletions tests/test_mode_detection.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
microservices = string type (ANDROID/IOS), monolith = int type code.
"""
import pytest
import requests
import responses

from mdast_cli_core.factory import (MODE_MICROSERVICES, MODE_MONOLITH, ModeDetectionError,
Expand Down Expand Up @@ -82,3 +83,85 @@ def test_autodetect_failure_network(mocked_responses, monkeypatch):
with pytest.raises(ModeDetectionError) as excinfo:
resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID)
assert excinfo.value.auth_error is False


# --- TLS-inspecting proxy: detection must not block a flow that does not verify ---

def test_tls_failure_falls_back_to_unverified_probe(mocked_responses, monkeypatch):
"""A TLS-inspecting proxy must not block a monolith run.

The monolith flow itself sends every request with verify=False, so failing the
detection probe on certificate verification blocks a door with no wall behind it.
"""
monkeypatch.delenv('MDAST_CLI_MODE', raising=False)
# Verified probes (Bearer, then Token) both fail the handshake...
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.SSLError(
'certificate verify failed: self-signed certificate in certificate chain'))
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.SSLError(
'certificate verify failed: self-signed certificate in certificate chain'))
# ...the unverified retry succeeds and classifies the stand.
mocked_responses.add(responses.GET, ARCH_URL, status=401)
mocked_responses.add(responses.GET, ARCH_URL, json=MONO_ARCH)
assert resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID) == MODE_MONOLITH
assert mocked_responses.calls[-1].request.req_kwargs['verify'] is False


def test_tls_fallback_does_not_silently_accept_microservices(mocked_responses, monkeypatch):
"""Microservices verify certificates for real traffic, so resolving the mode over
an unverified channel would only defer the failure - demand an explicit decision."""
monkeypatch.delenv('MDAST_CLI_MODE', raising=False)
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.SSLError('boom'))
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.SSLError('boom'))
mocked_responses.add(responses.GET, ARCH_URL, json=MS_ARCH)
with pytest.raises(ModeDetectionError) as excinfo:
resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID)
assert 'REQUESTS_CA_BUNDLE' in str(excinfo.value)


def test_no_tls_fallback_when_verification_already_disabled(mocked_responses, monkeypatch):
"""With verify=False there is nothing to fall back from - do not probe twice."""
monkeypatch.delenv('MDAST_CLI_MODE', raising=False)
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.SSLError('boom'))
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.SSLError('boom'))
with pytest.raises(ModeDetectionError):
resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID, verify=False)
assert len(mocked_responses.calls) == 2


# --- diagnosis: name the real cause instead of blaming --url/--token ---

def test_proxy_error_points_at_proxy_env_vars(mocked_responses, monkeypatch):
monkeypatch.delenv('MDAST_CLI_MODE', raising=False)
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.ProxyError(
'Unable to connect to proxy'))
mocked_responses.add(responses.GET, ARCH_URL, body=requests.exceptions.ProxyError(
'Unable to connect to proxy'))
with pytest.raises(ModeDetectionError) as excinfo:
resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID)
message = str(excinfo.value)
assert 'HTTPS_PROXY' in message
assert '--token' not in message


def test_auth_failure_quotes_server_explanation(mocked_responses, monkeypatch):
"""The server says 'Token has expired' - that must reach the operator."""
monkeypatch.delenv('MDAST_CLI_MODE', raising=False)
body = {'message': 'Token has expired', 'detail': [{'code': 900}]}
mocked_responses.add(responses.GET, ARCH_URL, json=body, status=401)
mocked_responses.add(responses.GET, ARCH_URL, json=body, status=401)
with pytest.raises(ModeDetectionError) as excinfo:
resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID)
assert 'Token has expired' in str(excinfo.value)
assert excinfo.value.auth_error is True


def test_quoted_body_never_leaks_the_token(mocked_responses, monkeypatch):
"""A server that echoes the credential back must not turn diagnostics into a leak."""
monkeypatch.delenv('MDAST_CLI_MODE', raising=False)
body = {'message': f'Invalid token: {TOKEN}'}
mocked_responses.add(responses.GET, ARCH_URL, json=body, status=401)
mocked_responses.add(responses.GET, ARCH_URL, json=body, status=401)
with pytest.raises(ModeDetectionError) as excinfo:
resolve_installation_mode(REST_URL, TOKEN, COMPANY_ID)
assert TOKEN not in str(excinfo.value)
assert '***' in str(excinfo.value)
Loading