Add sanitization and a test

Author: basepi

elasticapm/base.py

@@ -36,6 +36,7 @@
 import logging
 import os
 import platform
+import re
 import sys
 import threading
 import time
@@ -425,11 +426,10 @@ def get_user_agent(self) -> str:
  to the APM Server
  """
  if self.config.service_version:
- return "apm-agent-python/{} ({} {})".format(
- elasticapm.VERSION, self.config.service_name, self.config.service_version
- ).encode("utf-8")
+ service_version = re.sub(r"[^\t _\x21-\x27\x2a-\x5b\x5d-\x7e\x80-\xff]", "_", self.config.service_version)
+ return "apm-agent-python/{} ({} {})".format(elasticapm.VERSION, self.config.service_name, service_version)
  else:
- return "apm-agent-python/{} ({})".format(elasticapm.VERSION, self.config.service_name).encode("utf-8")
+ return "apm-agent-python/{} ({})".format(elasticapm.VERSION, self.config.service_name)
 
  def build_metadata(self):
  data = {

tests/client/client_tests.py

@@ -867,3 +867,15 @@ def test_backdating_transaction(elasticapm_client):
  elasticapm_client.end_transaction()
  transaction = elasticapm_client.events[TRANSACTION][0]
  assert 1000 < transaction["duration"] < 2000
+
+
+@pytest.mark.parametrize(
+ "elasticapm_client,expected",
+ [
+ ({"service_version": "v2"}, "v2"),
+ ({"service_version": "v2 \x00"}, "v2 _"),
+ ],
+ indirect=["elasticapm_client"],
+)
+def test_user_agent(elasticapm_client, expected):
+ assert elasticapm_client.get_user_agent() == "apm-agent-python/unknown (myapp {})".format(expected)