diff --git a/prometheus_client/exposition.py b/prometheus_client/exposition.py index 0b63f6f6..8dd62944 100644 --- a/prometheus_client/exposition.py +++ b/prometheus_client/exposition.py @@ -372,7 +372,7 @@ def choose_encoder(accept_header: str) -> Tuple[Callable[[Collector], bytes], st # mimetype. if not version: return (partial(openmetrics.generate_latest, escaping=openmetrics.UNDERSCORES, version="1.0.0"), openmetrics.CONTENT_TYPE_LATEST) - if version and parse_version(version) >= (1, 0, 0): + if version and _version_at_least(version, (1, 0, 0)): return (partial(openmetrics.generate_latest, escaping=escaping, version=version), f'application/openmetrics-text; version={version}; charset=utf-8; escaping=' + str(escaping)) elif accepted.split(';')[0].strip() == 'text/plain': @@ -381,12 +381,23 @@ def choose_encoder(accept_header: str) -> Tuple[Callable[[Collector], bytes], st escaping = _get_escaping(toks) # Only return an escaping header if we have a good version and # mimetype. - if version and parse_version(version) >= (1, 0, 0): + if version and _version_at_least(version, (1, 0, 0)): return (partial(generate_latest, escaping=escaping), CONTENT_TYPE_LATEST + '; escaping=' + str(escaping)) return generate_latest, CONTENT_TYPE_PLAIN_0_0_4 +def _version_at_least(version: str, minimum: Tuple[int, ...]) -> bool: + """Compare a version parameter from the Accept header. + + A version with a non-numeric part, such as 1.0.0-rc.1 or a quoted + value, is treated as unsupported.""" + parsed = parse_version(version) + if not all(isinstance(part, int) for part in parsed): + return False + return parsed >= minimum + + def _get_version(accept_header: List[str]) -> str: """Return the version tag from the Accept header. diff --git a/tests/test_exposition.py b/tests/test_exposition.py index 1885480f..b1acca44 100644 --- a/tests/test_exposition.py +++ b/tests/test_exposition.py @@ -635,6 +635,20 @@ def test_prom_no_version(self): self.assert_is_prom(exp) + def test_non_numeric_version(self): + for accept_header in ( + 'application/openmetrics-text; version=1.0.0-rc.1', + 'application/openmetrics-text; version="1.0.0"', + 'text/plain; version=latest', + ): + with self.subTest(accept_header=accept_header): + generator, content_type = choose_encoder(accept_header) + assert content_type == CONTENT_TYPE_PLAIN_0_0_4 + exp = generator(self.registry).decode('utf-8') + self.assert_is_escaped(exp) + self.assert_is_prom(exp) + + class TestWsgiTLS(unittest.TestCase): def setUp(self): self.certs_dir = os.path.join(