Remove private tokens from source error messages and collector logging

components/collector/src/base_collectors/source_collector.py

@@ -86,7 +86,7 @@ async def __safely_get_source_responses(self) -> Tuple[Responses, URL, ErrorMess
             responses = await self._get_source_responses(api_url := await self._api_url())
             logging.info("Retrieved %s", tokenless(api_url) or self.__class__.__name__)
         except aiohttp.ClientError as reason:
-            error = str(reason)
+            error = tokenless(str(reason))
             logging.warning("Failed to retrieve %s: %s", tokenless(api_url) or self.__class__.__name__, reason)
         except Exception as reason:  # pylint: disable=broad-except
             error = stable_traceback(traceback.format_exc())

components/collector/src/collector_utilities/functions.py

@@ -44,9 +44,9 @@ def stable_traceback(traceback: str) -> str:
     return traceback
 
 
-def tokenless(url: URL) -> URL:
-    """Strip private tokens from the url."""
-    return URL(re.sub(TOKEN_SUB[0], TOKEN_SUB[1], url))
+def tokenless(url: str) -> str:
+    """Strip private tokens from (text with) urls."""
+    return re.sub(TOKEN_SUB[0], TOKEN_SUB[1], url)
 
 
 def hashless(url: URL) -> URL:

components/server/src/model/queries.py

@@ -25,3 +25,11 @@ def get_metric_uuid(reports, source_uuid: SourceId) -> Optional[MetricId]:
         for subject in report["subjects"].values():
             metrics.extend(subject["metrics"].items())
     return [metric_uuid for (metric_uuid, metric) in metrics if source_uuid in metric["sources"]][0]
+
+
+def is_password_parameter(data_model, source_type: str, parameter: str) -> bool:
+    """Return whether the parameter of the source type is a password."""
+    # If the parameter key can't be found (this can happen when the parameter is removed from the data model),
+    # err on the safe side and assume it was a password type
+    parameter_type = data_model["sources"][source_type]["parameters"].get(parameter, dict(type="password"))["type"]
+    return str(parameter_type) == "password"

components/server/src/model/transformations.py

@@ -6,18 +6,15 @@
 from server_utilities.functions import unique
 from server_utilities.type import Color, EditScope, ItemId, Status
 from .iterators import sources as iter_sources
+from .queries import is_password_parameter
 
 
 def hide_credentials(data_model, *reports) -> None:
     """Hide the credentials in the reports."""
-    data_model_sources = data_model["sources"]
     for report in reports:
         for _, source in iter_sources(report):
             for parameter_key, parameter_value in source.get("parameters", {}).items():
-                # If the parameter key can't be found (this can happen when the parameter is removed from the data
-                # model), err on the safe side and assume it was a password type
-                if parameter_value and data_model_sources[source["type"]]["parameters"].get(
-                        parameter_key, dict(type="password"))["type"] == "password":
+                if parameter_value and is_password_parameter(data_model, source["type"], parameter_key):
                     source["parameters"][parameter_key] = "this string replaces credentials"
 
 

components/server/src/routes/source.py

@@ -11,6 +11,7 @@
 from database.reports import get_data, insert_new_report
 from model.actions import copy_source, move_item
 from model.transformations import change_source_parameter
+from model.queries import is_password_parameter
 from server_utilities.functions import uuid
 from server_utilities.type import EditScope, MetricId, SourceId, URL
 
@@ -125,7 +126,7 @@ def post_source_parameter(source_uuid: SourceId, parameter_key: str, database: D
     edit_scope = cast(EditScope, dict(bottle.request.json).get("edit_scope", "source"))
     changed_ids = change_source_parameter(data, parameter_key, old_value, new_value, edit_scope)
 
-    if data.datamodel["sources"][data.source["type"]]["parameters"][parameter_key]["type"] == "password":
+    if is_password_parameter(data.datamodel, data.source["type"], parameter_key):
         new_value, old_value = "*" * len(new_value), "*" * len(old_value)
 
     source_description = _source_description(data, edit_scope, parameter_key, old_value)
@@ -188,11 +189,11 @@ def _check_url_availability(url: URL, source_parameters: Dict[str, str]) -> Dict
 
 def _basic_auth_credentials(source_parameters) -> Optional[Tuple[str, str]]:
     """Return the basic authentication credentials, if any."""
-    if "private_token" in source_parameters:
-        return source_parameters["private_token"], ""
-    if "username" in source_parameters and "password" in source_parameters:
-        return source_parameters["username"], source_parameters["password"]
-    return None
+    if private_token := source_parameters.get("private_token", ""):
+        return private_token, ""
+    username = source_parameters.get("username", "")
+    password = source_parameters.get("password", "")
+    return (username, password) if username and password else None
 
 
 def _headers(source_parameters) -> Dict:

docs/CHANGELOG.md

@@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 - Make it clear in the user interface and the documentation that Quality-time can be authenticated with Jenkins using a username and API token, in addition to a username and password. Closes [#1125](https://github.com/ICTU/quality-time/issues/1125). 
 
+### Fixed
+
+- Remove private tokens from source error messages and collector logging. Fixes [#1127](https://github.com/ICTU/quality-time/issues/1127).
+
 ## [2.1.1] - [2020-04-03]
 
 ### Fixed