Fix security issues flagged in PR #593 code scanning.

Signed-off-by: lakshmj <[email protected]>

Author: lakshmj

examples/nslaslicense_offline.yaml

@@ -20,3 +20,22 @@
     - name: Display license result
       ansible.builtin.debug:
         var: lic_result
+
+    - name: Apply offline LAS license with host key checking disabled (trusted isolated environment)
+      delegate_to: localhost
+      netscaler.adc.nslaslicense_offline:
+        nsip: "{{ nsip }}"
+        nitro_user: "{{ nitro_user }}"
+        nitro_pass: "{{ nitro_pass }}"
+        nitro_protocol: "{{ nitro_protocol | default('https') }}"
+        validate_certs: false
+        host_key_checking: false
+        request_pem: CNS_V10000_SERVER
+        request_ed: Premium
+        is_fips: false
+        las_secrets_json: /path/to/las_secrets.json
+      register: lic_result_no_hkc
+
+    - name: Display license result (no host key checking)
+      ansible.builtin.debug:
+        var: lic_result_no_hkc

plugins/module_utils/las_utils.py

@@ -11,6 +11,7 @@
 import os
 import re
 import subprocess
+import tempfile
 import uuid
 
 from ansible.module_utils.urls import open_url
@@ -207,7 +208,8 @@ def build_multipart(fields, files):
 class LASClient:
     """Client for the LAS (License Activation Service) cloud API."""
 
-    _BEARER_CACHE = "/tmp/r56_bearer"
+    # Namespaced by effective user ID to avoid insecure shared /tmp file access.
+    _BEARER_CACHE = os.path.join(tempfile.gettempdir(), "r56_bearer_{0}".format(os.geteuid()))
 
     def __init__(self, lsguid, secret_file):
         self.endpoint = "netscalerfixedbw"
@@ -317,9 +319,16 @@ def get_blob_from_las(self, newactivationid, lsfingerprint, output_file, bearer,
 # ---------------------------------------------------------------------------
 
 
-def sftp_get(ip, username, password, remote_path, local_path, loglines):
+def sftp_get(ip, username, password, remote_path, local_path, loglines, host_key_checking=True):
     ssh = paramiko.SSHClient()
-    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    ssh.load_system_host_keys()
+    if host_key_checking:
+        # RejectPolicy raises an error for unknown host keys, preventing silent MITM attacks.
+        # The ADC device's SSH host key must be present in the control node's known_hosts.
+        ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
+    else:
+        # User explicitly opted out of host key checking (host_key_checking=false).
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
     sftp = None
     try:
         ssh.connect(ip, username=username, password=password)
@@ -334,9 +343,16 @@ def sftp_get(ip, username, password, remote_path, local_path, loglines):
         ssh.close()
 
 
-def sftp_put(ip, username, password, local_path, remote_path, loglines):
+def sftp_put(ip, username, password, local_path, remote_path, loglines, host_key_checking=True):
     ssh = paramiko.SSHClient()
-    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+    ssh.load_system_host_keys()
+    if host_key_checking:
+        # RejectPolicy raises an error for unknown host keys, preventing silent MITM attacks.
+        # The ADC device's SSH host key must be present in the control node's known_hosts.
+        ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
+    else:
+        # User explicitly opted out of host key checking (host_key_checking=false).
+        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
     sftp = None
     try:
         ssh.connect(ip, port=22, username=username, password=password)
@@ -427,7 +443,7 @@ def check_if_new_api(mapping, release, major, minor):
 # ---------------------------------------------------------------------------
 
 
-def get_offline_request_package(nitro, ip, username, password, local_dir, new_api, loglines):
+def get_offline_request_package(nitro, ip, username, password, local_dir, new_api, loglines, host_key_checking=True):
     """Trigger NITRO to generate the NS offline activation request tgz, then SFTP it to local_dir."""
     resource = "nslicenseactivationdata?args=usehostname:true" if new_api else "nslicenseactivationdata"
     o = nitro.get(resource)
@@ -438,7 +454,7 @@ def get_offline_request_package(nitro, ip, username, password, local_dir, new_ap
         return ""
 
     local_path = os.path.join(local_dir, src_file)
-    sftp_get(ip, username, password, "/nsconfig/license/" + src_file, local_path, loglines)
+    sftp_get(ip, username, password, "/nsconfig/license/" + src_file, local_path, loglines, host_key_checking)
     return src_file
 
 
@@ -449,7 +465,13 @@ def get_offline_request_package(nitro, ip, username, password, local_dir, new_ap
 
 def extract_lsguid(file_path, loglines):
     dest_dir = os.path.dirname(file_path)
+    # Validate that file_path is within dest_dir to guard against path traversal.
+    real_file_path = os.path.realpath(file_path)
+    real_dest_dir = os.path.realpath(dest_dir)
+    if not real_file_path.startswith(real_dest_dir + os.sep):
+        raise RuntimeError("Invalid file path outside temp directory: {0}".format(file_path))
     json_file = "ns_offline_activation_request.json"
+    # shell=False ensures no shell metacharacter interpretation; all args are controlled internally.
     cmd = [
         "tar",
         "-xvf",
@@ -479,12 +501,12 @@ def extract_lsguid(file_path, loglines):
 
     try:
         os.remove(json_path)
-    except Exception:
-        pass
+    except Exception as e:
+        loglines.append("DEBUG: Could not remove temp file {0}: {1}".format(json_path, str(e)))
     try:
         os.remove(os.path.join(dest_dir, "lasData.tgz"))
-    except Exception:
-        pass
+    except Exception as e:
+        loglines.append("DEBUG: Could not remove temp file lasData.tgz: {0}".format(str(e)))
 
     lsguid = data["lsguid"]
     loglines.append("INFO: Extracted lsguid: {0}".format(lsguid))
@@ -496,8 +518,8 @@ def extract_lsguid(file_path, loglines):
 # ---------------------------------------------------------------------------
 
 
-def apply_license_blob_ns(nitro, ip, username, password, fname, loglines):
-    sftp_put(ip, username, password, fname, "/nsconfig/license/" + fname, loglines)
+def apply_license_blob_ns(nitro, ip, username, password, fname, loglines, host_key_checking=True):
+    sftp_put(ip, username, password, fname, "/nsconfig/license/" + fname, loglines, host_key_checking)
     payload = {
         "params": {"action": "apply", "warning": "YES"},
         "nslaslicense": {"filename": fname, "filelocation": "/nsconfig/license", "fixedbandwidth": True},

plugins/modules/nslaslicense_offline.py

@@ -89,6 +89,14 @@
       - The file must contain the keys C(ccid), C(client), C(password), C(las_endpoint), and C(cc_endpoint).
     type: str
     required: true
+  host_key_checking:
+    description:
+      - If C(false), the ADC device's SSH host key will not be validated during SFTP transfers.
+      - Disabling this is insecure and should only be used in trusted, isolated environments.
+      - When C(true) (default), the device's SSH host key must be present in the control node's known_hosts file.
+        Use C(ssh-keyscan <nsip> >> ~/.ssh/known_hosts) to add it.
+    type: bool
+    default: true
 """
 
 EXAMPLES = r"""
@@ -127,6 +135,18 @@
     request_ed: Standard
     las_secrets_json: /etc/netscaler/zmcd_secrets.json
 
+- name: Apply offline LAS license with host key checking disabled (trusted isolated environment)
+  delegate_to: localhost
+  netscaler.adc.nslaslicense_offline:
+    nsip: 10.102.201.230
+    nitro_user: nsroot
+    nitro_pass: "{{ nitro_pass }}"
+    validate_certs: false
+    host_key_checking: false
+    request_pem: CNS_8905_SERVER
+    request_ed: Premium
+    las_secrets_json: /etc/netscaler/zmcd_secrets.json
+
 """
 
 RETURN = r"""
@@ -194,6 +214,7 @@ def main():
         request_ed=dict(required=True, type="str", choices=["Advanced", "Premium", "Standard"]),
         is_fips=dict(type="bool", default=False),
         las_secrets_json=dict(required=True, type="str"),
+        host_key_checking=dict(type="bool", default=True),
     )
 
     module = AnsibleModule(
@@ -214,6 +235,7 @@ def main():
     request_ed = module.params["request_ed"]
     is_fips = module.params["is_fips"]
     las_secrets_json = module.params["las_secrets_json"]
+    host_key_checking = module.params["host_key_checking"]
     if username != "nsroot":
         module.fail_json(msg="Only the 'nsroot' account is supported. Got: '{0}'".format(username), **result)
 
@@ -252,7 +274,7 @@ def main():
     # Get activation request package from device
     temp_dir = os.path.join(tempfile.mkdtemp(prefix="nslas_"), "")
     try:
-        ns_file_name = get_offline_request_package(nitro, ip, username, password, temp_dir, new_api, loglines)
+        ns_file_name = get_offline_request_package(nitro, ip, username, password, temp_dir, new_api, loglines, host_key_checking)
         if not ns_file_name:
             module.fail_json(msg="Failed to retrieve activation request package from device", **result)
 
@@ -264,7 +286,7 @@ def main():
             lsguid = extract_lsguid(request_file, loglines)
         except Exception as e:
             loglines.append("WARNING: First parse attempt failed ({0}), re-downloading package".format(str(e)))
-            ns_file_name = get_offline_request_package(nitro, ip, username, password, temp_dir, new_api, loglines)
+            ns_file_name = get_offline_request_package(nitro, ip, username, password, temp_dir, new_api, loglines, host_key_checking)
             if not ns_file_name:
                 module.fail_json(msg="Re-download of activation request package failed", **result)
             request_file = os.path.join(temp_dir, ns_file_name)
@@ -276,7 +298,7 @@ def main():
             module.fail_json(msg="Failed to generate offline license token from LAS", **result)
 
         # Apply license blob to device
-        apply_license_blob_ns(nitro, ip, username, password, output_file, loglines)
+        apply_license_blob_ns(nitro, ip, username, password, output_file, loglines, host_key_checking)
 
         result["changed"] = True
         result["output_file"] = output_file