Mask Password in Log table when using the CLI

airflow/utils/cli.py

@@ -105,8 +105,15 @@ def _build_metrics(func_name, namespace):
     :param namespace: Namespace instance from argparse
     :return: dict with metrics
     """
+    sensitive_fields = {'-p', '--password'}
+    full_command = list(sys.argv)
+    for sensitive_field in sensitive_fields:
+        if sensitive_field in full_command:
+            # Mask value passed to Password arg
+            full_command[full_command.index(sensitive_field) + 1] = "*" * 8
+
     metrics = {'sub_command': func_name, 'start_datetime': datetime.utcnow(),
-               'full_command': '{}'.format(list(sys.argv)), 'user': getpass.getuser()}
+               'full_command': '{}'.format(full_command), 'user': getpass.getuser()}
 
     if not isinstance(namespace, Namespace):
         raise ValueError("namespace argument should be argparse.Namespace instance,"

tests/utils/test_cli_util.py

@@ -16,12 +16,16 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-
+import json
 import os
+import sys
 import unittest
 from argparse import Namespace
 from contextlib import contextmanager
 from datetime import datetime
+from unittest import mock
+
+from parameterized import parameterized
 
 from airflow import settings
 from airflow.exceptions import AirflowException
@@ -85,6 +89,34 @@ def test_get_dags(self):
         with self.assertRaises(AirflowException):
             cli.get_dags(None, "foobar", True)
 
+    @parameterized.expand(["--password", "-p"])
+    def test_cli_create_user_supplied_password_is_masked(self, password_variant):
+
+        args = [
+            'airflow', 'users', 'create', '--username', 'test2', '--lastname', 'doe',
+            '--firstname', 'jon',
+            '--email', '[email protected]', '--role', 'Viewer', password_variant, 'test'
+        ]
+
+        expected_command = [
+            'airflow', 'users', 'create', '--username', 'test2', '--lastname', 'doe',
+            '--firstname', 'jon',
+            '--email', '[email protected]', '--role', 'Viewer', password_variant, '********'
+        ]
+
+        exec_date = datetime.utcnow()
+        namespace = Namespace(dag_id='foo', task_id='bar', subcommand='test', execution_date=exec_date)
+        with mock.patch.object(sys, "argv", args):
+            metrics = cli._build_metrics(args[1], namespace)
+
+        self.assertTrue(metrics.get('start_datetime') <= datetime.utcnow())
+
+        log = metrics.get('log')
+        command = json.loads(log.extra).get('full_command')  # type: str
+        # Replace single quotes to double quotes to avoid json decode error
+        command = json.loads(command.replace("'", '"'))
+        self.assertEqual(command, expected_command)
+
 
 @contextmanager
 def fail_action_logger_callback():