[WIP] Configure bandit to medium and fix security issues

.pre-commit-config.yaml

@@ -993,9 +993,9 @@ repos:
           airflow/example_dags/.*
         args:
           - "--skip"
-          - "B301,B324,B403,B404,B603"
+          - "B108,B113,B301,B324,B403,B404,B603,B604"
           - "--severity-level"
-          - "high"  # TODO: remove this line when we fix all the issues
+          - "medium"  # TODO: remove this line when we fix all the issues
         ## ADD MOST PRE-COMMITS ABOVE THAT LINE
         # The below pre-commits are those requiring CI image to be built
       - id: mypy-dev

airflow/decorators/__init__.pyi

@@ -338,7 +338,7 @@ class TaskDecoratorCollection:
         tls_hostname: str | bool | None = None,
         tls_ssl_version: str | None = None,
         mount_tmp_dir: bool = True,
-        tmp_dir: str = "/tmp/airflow",
+        tmp_dir: str = "/tmp/airflow",  # nosec
         user: str | int | None = None,
         mounts: list[str] | None = None,
         entrypoint: str | list[str] | None = None,

airflow/providers/docker/operators/docker.py

@@ -211,7 +211,7 @@ def __init__(
         tls_hostname: str | bool | None = None,
         tls_ssl_version: str | None = None,
         mount_tmp_dir: bool = True,
-        tmp_dir: str = "/tmp/airflow",
+        tmp_dir: str = "/tmp/airflow",  # nosec
         user: str | int | None = None,
         mounts: list[Mount] | None = None,
         entrypoint: str | list[str] | None = None,

airflow/providers/google/cloud/example_dags/example_cloud_sql_query.py

@@ -55,7 +55,7 @@
 GCSQL_POSTGRES_DATABASE_NAME = os.environ.get("GCSQL_POSTGRES_DATABASE_NAME", "postgresdb")
 GCSQL_POSTGRES_USER = os.environ.get("GCSQL_POSTGRES_USER", "postgres_user")
 GCSQL_POSTGRES_PASSWORD = os.environ.get("GCSQL_POSTGRES_PASSWORD", "JoxHlwrPzwch0gz9")
-GCSQL_POSTGRES_PUBLIC_IP = os.environ.get("GCSQL_POSTGRES_PUBLIC_IP", "0.0.0.0")
+GCSQL_POSTGRES_PUBLIC_IP = os.environ.get("GCSQL_POSTGRES_PUBLIC_IP", "0.0.0.0")  # nosec
 GCSQL_POSTGRES_PUBLIC_PORT = os.environ.get("GCSQL_POSTGRES_PUBLIC_PORT", 5432)
 GCSQL_POSTGRES_CLIENT_CERT_FILE = os.environ.get(
     "GCSQL_POSTGRES_CLIENT_CERT_FILE", ".key/postgres-client-cert.pem"
@@ -69,7 +69,7 @@
 GCSQL_MYSQL_DATABASE_NAME = os.environ.get("GCSQL_MYSQL_DATABASE_NAME", "mysqldb")
 GCSQL_MYSQL_USER = os.environ.get("GCSQL_MYSQL_USER", "mysql_user")
 GCSQL_MYSQL_PASSWORD = os.environ.get("GCSQL_MYSQL_PASSWORD", "JoxHlwrPzwch0gz9")
-GCSQL_MYSQL_PUBLIC_IP = os.environ.get("GCSQL_MYSQL_PUBLIC_IP", "0.0.0.0")
+GCSQL_MYSQL_PUBLIC_IP = os.environ.get("GCSQL_MYSQL_PUBLIC_IP", "0.0.0.0")  # nosec
 GCSQL_MYSQL_PUBLIC_PORT = os.environ.get("GCSQL_MYSQL_PUBLIC_PORT", 3306)
 GCSQL_MYSQL_CLIENT_CERT_FILE = os.environ.get("GCSQL_MYSQL_CLIENT_CERT_FILE", ".key/mysql-client-cert.pem")
 GCSQL_MYSQL_CLIENT_KEY_FILE = os.environ.get("GCSQL_MYSQL_CLIENT_KEY_FILE", ".key/mysql-client-key.pem")
@@ -193,7 +193,7 @@ def get_absolute_path(path):
 try:
     sql_proxy_binary_path = subprocess.check_output(["which", "cloud_sql_proxy"]).decode("utf-8").rstrip()
 except subprocess.CalledProcessError:
-    sql_proxy_binary_path = "/tmp/anyhow_download_cloud_sql_proxy"
+    sql_proxy_binary_path = "/tmp/anyhow_download_cloud_sql_proxy"  # nosec
 
 os.environ["AIRFLOW_CONN_PROXY_MYSQL_SOCKET"] = (
     "gcpcloudsql://{user}:{password}@{public_ip}:{public_port}/{database}?"

airflow/providers/google/leveldb/hooks/leveldb.py

@@ -51,7 +51,7 @@ def __init__(self, leveldb_conn_id: str = default_conn_name):
         self.connection = self.get_connection(leveldb_conn_id)
         self.db: plyvel.DB | None = None
 
-    def get_conn(self, name: str = "/tmp/testdb/", create_if_missing: bool = False, **kwargs) -> DB:
+    def get_conn(self, name: str = "/tmp/testdb/", create_if_missing: bool = False, **kwargs) -> DB:  # nosec
         """
         Creates `Plyvel DB <https://plyvel.readthedocs.io/en/latest/api.html#DB>`__.
 

airflow/providers/google/leveldb/operators/leveldb.py

@@ -54,7 +54,7 @@ def __init__(
         keys: list[bytes] | None = None,
         values: list[bytes] | None = None,
         leveldb_conn_id: str = "leveldb_default",
-        name: str = "/tmp/testdb/",
+        name: str = "/tmp/testdb/",  # nosec
         create_if_missing: bool = True,
         create_db_extra_options: dict[str, Any] | None = None,
         **kwargs,

airflow/providers/ssh/hooks/ssh.py

@@ -488,7 +488,7 @@ def exec_ssh_client_command(
 
         # set timeout taken as params
         stdin, stdout, stderr = ssh_client.exec_command(
-            command=command,
+            command=command,  # nosec
             get_pty=get_pty,
             timeout=cmd_timeout,
             environment=environment,
@@ -556,7 +556,7 @@ def test_connection(self) -> tuple[bool, str]:
         """Test the ssh connection by execute remote bash commands."""
         try:
             with self.get_conn() as conn:
-                conn.exec_command("pwd")
+                conn.exec_command("pwd")  # nosec
             return True, "Connection successfully tested"
         except Exception as e:
             return False, str(e)

airflow/security/utils.py

@@ -56,7 +56,7 @@ def get_components(principal) -> list[str] | None:
 def replace_hostname_pattern(components, host=None):
     """Replace hostname with the right pattern including lowercase of the name."""
     fqdn = host
-    if not fqdn or fqdn == "0.0.0.0":
+    if not fqdn or fqdn == "0.0.0.0":  # nosec
         fqdn = get_hostname()
     return f"{components[0]}/{fqdn.lower()}@{components[2]}"
 

airflow/utils/net.py

@@ -32,7 +32,7 @@ def getfqdn(name=""):
     An empty argument is interpreted as meaning the local host.
     """
     name = name.strip()
-    if not name or name == "0.0.0.0":
+    if not name or name == "0.0.0.0":  # nosec
         name = socket.gethostname()
     try:
         addrs = socket.getaddrinfo(name, None, 0, socket.SOCK_DGRAM, 0, socket.AI_CANONNAME)

tests/providers/amazon/aws/hooks/test_base_aws.py

@@ -664,15 +664,15 @@ def test_assume_role_with_saml(self, mock_get_connection, sts_endpoint):
         mock_requests_gssapi = mock.Mock()
         mock_auth = mock_requests_gssapi.HTTPSPNEGOAuth()
 
-        mock_lxml = mock.Mock()
-        mock_xpath = mock_lxml.etree.fromstring.return_value.xpath
+        mock_defusedxml = mock.Mock()
+        mock_xpath = mock_defusedxml.ElementTree.fromstring.return_value.xpath
         mock_xpath.return_value = encoded_saml_assertion
 
         def import_mock(name, *args, **kwargs):
             if name == "requests_gssapi":
                 return mock_requests_gssapi
-            if name == "lxml":
-                return mock_lxml
+            if name == "defusedxml":
+                return mock_defusedxml
             return orig_import(name, *args, **kwargs)
 
         def mock_assume_role_with_saml(**kwargs):