fix(codeql): batch-4 medium-sev + warnings + log-injection defense

docs/evaluator/test_doc_examples.py

@@ -53,6 +53,7 @@
 """
 
 import time
+
 import httpx
 
 BASE_URL = "http://localhost:8080"
@@ -161,7 +162,7 @@ def main():
     print(f"Results: {passed} passed, {failed} failed")
     print("=" * 60)
 
-    exit(1 if failed else 0)
+    raise SystemExit(1 if failed else 0)
 
 
 if __name__ == "__main__":

docs/javascripts/api-filter.js

@@ -9,7 +9,7 @@
     var iframe = document.querySelector('iframe.swagger-ui-iframe');
     if (!chipContainer || !iframe) return;
     var hiddenTags = {};
-    if (chipContainer && chipContainer.dataset.hiddenTags) {
+    if (chipContainer.dataset.hiddenTags) {
       chipContainer.dataset.hiddenTags.split(',').forEach(function(tag) {
         tag = tag.trim();
         if (tag) hiddenTags[tag] = true;

packages/nemo_evaluator_sdk/tests/execution/test_benchmark_execution.py

@@ -107,9 +107,8 @@ def output_spec(self) -> list[MetricOutputSpec]:
 class _NoOpProgressReporter:
     """Progress reporter stub for structural protocol checks."""
 
-    def increment_work(self, increment: int = 1, /) -> None:
+    def increment_work(self, increment: int = 1, /) -> None:  # noqa: ARG002
         """Accept progress increments without side effects."""
-        del increment
 
 
 class _CorpusMetric(_ScriptedMetric):

packages/nemo_nb/nemo_nb/sphinx.py

@@ -109,7 +109,6 @@ def should_process_markdown(md_path: Path) -> bool:
             except Exception as e:
                 # If frontmatter parsing fails, continue to other checks
                 logger.debug(f"Frontmatter parsing failed for {md_path}: {e}")
-                pass
 
         return False
 

packages/nemo_nb/nemo_nb/sugar.py

@@ -132,7 +132,6 @@ def strip_leading_empty_lines(self, lines: List[str]) -> List[str]:
 
     def strip_trailing_empty_lines(self, lines: List[str]) -> List[str]:
         """Remove trailing empty lines from a list of lines."""
-        end = len(lines)
         for i in range(len(lines) - 1, -1, -1):
             if lines[i].strip():
                 end = i + 1

packages/nemo_platform_ext/src/nemo_platform_ext/quickstart/preflight.py

@@ -194,10 +194,10 @@ def _check_port_available(self) -> None:
         """Check if the configured host port is available or NeMo Platform is already running."""
         from .validators import is_quickstart_running
 
-        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         try:
-            sock.bind(("0.0.0.0", self.config.host_port))
-            sock.close()
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                sock.bind(("0.0.0.0", self.config.host_port))  # noqa: S104  # nosec B104
             self.results.append(
                 PreflightResult(
                     name="Port Available",

packages/nemo_platform_ext/src/nemo_platform_ext/quickstart/validators.py

@@ -214,10 +214,10 @@ def validate_port_available(port: int, config: QuickstartConfig | None = None) -
     """
     import socket
 
-    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     try:
-        sock.bind(("0.0.0.0", port))
-        sock.close()
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            sock.bind(("0.0.0.0", port))  # noqa: S104  # nosec B104
         return ValidationResult(True, f"Port {port} is available")
     except OSError:
         # Port is in use - check if it's the quickstart container

packages/nemo_platform_ext/tests/config/test_config.py

@@ -454,7 +454,8 @@ def test_invalid_context_name(self, temp_config_file: Path):
     def test_missing_cluster_reference(self, temp_config_file: Path):
         """Test error when context references missing cluster."""
         # Modify file to have invalid cluster reference
-        config_data = yaml.safe_load(open(temp_config_file))
+        with open(temp_config_file) as f:
+            config_data = yaml.safe_load(f)
         config_data["contexts"][0]["cluster"] = "nonexistent-cluster"
         with open(temp_config_file, "w") as f:
             yaml.dump(config_data, f)

packages/nemo_platform_plugin/src/nemo_platform_plugin/jobs/api_factory.py

@@ -349,7 +349,6 @@ async def _validated_lines_iterator(self, file_path: str, limit: int | None = No
                                 "error": {
                                     "type": e.__class__.__name__,
                                     "line": line_number,
-                                    "message": str(e),
                                 }
                             }
                         )

packages/nemo_platform_plugin/tests/test_commands.py

@@ -836,9 +836,8 @@ def _client_factory(*args, **kwargs):
     def test_submit_uses_nmp_base_url_env_when_no_flags(self, monkeypatch) -> None:
         captured_url: list[str] = []
 
-        def _fake_post(url: str, body: dict, *, headers: dict, timeout: float = 30.0, **_kwargs) -> None:
+        def _fake_post(url: str, body: dict, *, headers: dict, timeout: float = 30.0, **_kwargs) -> None:  # noqa: ARG001
             captured_url.append(url)
-            del body, headers, timeout
 
         monkeypatch.setattr("nemo_platform_plugin.commands._post_function_submit", _fake_post)
         monkeypatch.setenv("NMP_BASE_URL", "http://from-env:1234")

packages/nmp_common/src/nmp/common/observability/structured_logging.py

@@ -70,6 +70,25 @@ def _drop_console_hidden_fields(logger: logging.Logger, method_name: str, event_
     return event_dict
 
 
+# CR, LF, NEL, line/paragraph separators. Anything that a multi-line log
+# rendering pass could turn into a fake new log entry.
+_LOG_NEWLINE_CHARS = "\n\r\x0b\x0c\x1c\x1d\x1e\x85\u2028\u2029"
+_LOG_NEWLINE_TABLE = {ord(c): " " for c in _LOG_NEWLINE_CHARS}
+
+
+def _sanitize_log_strings(logger: logging.Logger, method_name: str, event_dict: EventDict) -> EventDict:
+    """Strip newline-like characters from string values in the event dict.
+
+    Defends against log-forgery via user-controlled fields landing in
+    request/response logs. JSON renderers escape these characters already,
+    but plain-text renderers do not.
+    """
+    for key, value in list(event_dict.items()):
+        if isinstance(value, str) and any(ch in value for ch in _LOG_NEWLINE_CHARS):
+            event_dict[key] = value.translate(_LOG_NEWLINE_TABLE)
+    return event_dict
+
+
 def clear_loggers():
     logging.getLogger().handlers.clear()
 
@@ -171,6 +190,7 @@ def _stamper(event_dict: EventDict) -> EventDict:
         structlog.processors.StackInfoRenderer(),
         structlog.processors.format_exc_info,
         structlog.processors.UnicodeDecoder(),
+        _sanitize_log_strings,
         structlog.processors.CallsiteParameterAdder(
             {
                 structlog.processors.CallsiteParameter.FILENAME,

packages/nmp_common/tests/observability/test_structured_logging.py

@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for the structured-logging processors."""
+
+from __future__ import annotations
+
+import logging
+
+import pytest
+from nmp.common.observability.structured_logging import _sanitize_log_strings
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("admin\n[ERROR] forged log line", "admin [ERROR] forged log line"),
+        ("with\r\ncrlf", "with  crlf"),
+        ("tab\tis fine", "tab\tis fine"),
+        ("plain string", "plain string"),
+        ("nel\x85next", "nel next"),
+        ("ls\u2028lsep", "ls lsep"),
+        ("ps\u2029psep", "ps psep"),
+    ],
+)
+def test_sanitize_log_strings_replaces_newline_variants(raw: str, expected: str) -> None:
+    event = {"event": "test", "user": raw}
+    result = _sanitize_log_strings(logging.getLogger(), "info", event)
+    assert result["user"] == expected
+
+
+def test_sanitize_log_strings_leaves_non_string_values_alone() -> None:
+    event = {"event": "test", "count": 5, "ok": True, "items": [1, 2, 3]}
+    result = _sanitize_log_strings(logging.getLogger(), "info", event)
+    assert result == event
+
+
+def test_sanitize_log_strings_skips_clean_strings() -> None:
+    event = {"event": "test", "name": "default/workspace"}
+    result = _sanitize_log_strings(logging.getLogger(), "info", event)
+    assert result["name"] == "default/workspace"
+
+
+def test_sanitize_log_strings_sanitizes_event_key() -> None:
+    """The 'event' key carries the primary log message — must be sanitized too."""
+    event = {"event": "user input was: bad\ninjected line"}
+    result = _sanitize_log_strings(logging.getLogger(), "info", event)
+    assert result["event"] == "user input was: bad injected line"
+
+
+def test_sanitize_log_strings_sanitizes_exception_field() -> None:
+    """structlog.format_exc_info writes a multi-line traceback into 'exception'.
+
+    The sanitizer must run after format_exc_info so attacker-controlled
+    exception messages can't forge log entries.
+    """
+    event = {"event": "boom", "exception": "Traceback (most recent call last):\n  ...\nValueError: pwn"}
+    result = _sanitize_log_strings(logging.getLogger(), "info", event)
+    assert "\n" not in result["exception"]
+
+
+def test_initialize_logging_wires_sanitizer_into_chain() -> None:
+    """Regression guard: 177 dismissed py/log-injection alerts depend on the sanitizer being in the chain."""
+    import logging as stdlib_logging
+
+    import structlog
+    from nmp.common.observability.structured_logging import _sanitize_log_strings, initialize_logging
+
+    root = stdlib_logging.getLogger()
+    saved_handlers = list(root.handlers)
+    saved_level = root.level
+    try:
+        root.handlers.clear()
+        initialize_logging()
+        # Locate the ProcessorFormatter by type, not handler index, so a future
+        # refactor that reorders or adds handlers doesn't silently false-pass.
+        formatters = [
+            h.formatter for h in root.handlers if isinstance(h.formatter, structlog.stdlib.ProcessorFormatter)
+        ]
+        assert formatters, "initialize_logging() did not attach a structlog ProcessorFormatter"
+        chain = getattr(formatters[0], "foreign_pre_chain", None) or []
+        assert _sanitize_log_strings in chain, (
+            "_sanitize_log_strings missing from structlog chain — log-injection defense is disabled"
+        )
+    finally:
+        root.handlers.clear()
+        root.handlers.extend(saved_handlers)
+        root.setLevel(saved_level)

plugins/nemo-auditor/tests/test_sdk_resources.py

@@ -281,8 +281,10 @@ def test_configs_and_targets_properties_are_cached() -> None:
     platform = _SyncPlatform()
     resource = AuditorPluginResource(cast(NeMoPlatform, platform))
 
-    assert resource.configs is resource.configs
-    assert resource.targets is resource.targets
+    cached_configs = resource.configs
+    cached_targets = resource.targets
+    assert resource.configs is cached_configs
+    assert resource.targets is cached_targets
 
 
 # ---------------------------------------------------------------------------

script/copyright_fixer.py

@@ -333,7 +333,8 @@ def _has_proprietary_license(head: str) -> bool:
 def _fix_proprietary_license(filepath: str) -> bool:
     """Replace LicenseRef-NvidiaProprietary with Apache-2.0. Returns True if modified."""
     try:
-        content = open(filepath, "r", encoding="utf-8").read()  # noqa: SIM115
+        with open(filepath, "r", encoding="utf-8") as f:
+            content = f.read()
     except (OSError, UnicodeDecodeError):
         return False
 
@@ -349,7 +350,8 @@ def _fix_proprietary_license(filepath: str) -> bool:
 def _fix_header_style(filepath: str) -> bool:
     """Fix wrong comment style on existing headers. Returns True if modified."""
     try:
-        content = open(filepath, "r", encoding="utf-8").read()  # noqa: SIM115
+        with open(filepath, "r", encoding="utf-8") as f:
+            content = f.read()
     except (OSError, UnicodeDecodeError):
         return False
 
@@ -375,7 +377,8 @@ def _fix_header_style(filepath: str) -> bool:
 def _fix_non_spdx_header(filepath: str) -> bool:
     """Replace legacy / non-standard copyright headers with correct SPDX. Returns True if modified."""
     try:
-        content = open(filepath, "r", encoding="utf-8").read()  # noqa: SIM115
+        with open(filepath, "r", encoding="utf-8") as f:
+            content = f.read()
     except (OSError, UnicodeDecodeError):
         return False
 
@@ -429,7 +432,8 @@ def _fix_non_spdx_header(filepath: str) -> bool:
 def _add_header(filepath: str) -> bool:
     """Add the copyright header to *filepath*. Returns True if modified."""
     try:
-        content = open(filepath, "r", encoding="utf-8").read()  # noqa: SIM115
+        with open(filepath, "r", encoding="utf-8") as f:
+            content = f.read()
     except (OSError, UnicodeDecodeError):
         return False
 

services/core/files/src/nmp/core/files/app/file_lock.py

@@ -127,7 +127,6 @@ async def _try_acquire(self, path: str, max_attempts: int = 3) -> bool:
             except EntityConflictError:
                 # Version changed - someone else took the lock, retry on next iteration
                 logger.debug("Update failed, lock was modified by another request")
-                pass
             except EntityNotFoundError:
                 pass  # Someone else deleted it, that's fine
 

services/core/jobs/tests/controllers/test_subprocess_backend.py

@@ -106,10 +106,12 @@ def test_schedule_uses_allowlisted_host_environment(mock_nmp_client, tmp_path, m
         [
             "/bin/sh",
             "-c",
-            'test "$PATH" = "/bin" && '
-            'test "$VIRTUAL_ENV" = "/venv" && '
-            'test -z "${HOME+x}" && '
-            'test -z "${SECRET_TOKEN+x}"',
+            (
+                'test "$PATH" = "/bin" && '
+                'test "$VIRTUAL_ENV" = "/venv" && '
+                'test -z "${HOME+x}" && '
+                'test -z "${SECRET_TOKEN+x}"'
+            ),
         ],
     )
 

services/core/models/src/nmp/core/models/controllers/backends/docker/creation_reconciler.py

@@ -292,7 +292,7 @@ def _is_port_free(self, port: int) -> bool:
         try:
             with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-                s.bind(("", port))
+                s.bind(("0.0.0.0", port))  # noqa: S104  # nosec B104
                 return True
         except OSError:
             logger.debug(f"Port {port} is not free (system process may be using it)")

services/core/models/src/nmp/core/models/controllers/provider_reconciler.py

@@ -836,8 +836,9 @@ def _generate_deployment_served_model_mappings(
                 served.append(ServedModelMapping(model_entity_id=model_entity_id, served_model_name=mid))
                 continue
 
-            # Prompt-tuned: parent is null, root equals base, id != base
-            if parent is None and root == base_id and mid != base_id:
+            # Prompt-tuned: parent is null, root equals base, id != base (the
+            # ``mid == base_id`` case was already handled by the base branch above).
+            if parent is None and root == base_id:
                 model_entity_id = f"{workspace}/{mid.removeprefix(f'{workspace}/')}"
                 served.append(ServedModelMapping(model_entity_id=model_entity_id, served_model_name=mid))
                 continue

services/core/models/src/nmp/core/models/parallelism/hueristics.py

@@ -646,8 +646,10 @@ def _compute_balance_bonus(
         elif ratio <= balance_cfg.ratio_good:
             return balance_cfg.bonus_good_very_large  # Good balance
 
-    # Large models: moderate balance preference
-    if param_count_b > size_thresholds.large or (num_experts > 0 and ep >= balance_cfg.ep_significant_threshold):
+    # Large models: moderate balance preference. By construction MoE models
+    # (num_experts > 0) already returned above, so only the dense check needs
+    # to fire here.
+    if param_count_b > size_thresholds.large:
         if ratio == balance_cfg.ratio_perfect:
             return balance_cfg.bonus_perfect_large  # Perfect balance
         elif ratio <= balance_cfg.ratio_good:

services/customizer/src/nmp/customizer/tasks/training/runner.py

@@ -150,7 +150,7 @@ def run(self) -> TrainingResult:
         finally:
             # === Phase 5: Write result (coordinator only) ===
             self._write_result(result)
-            return result
+        return result
 
     # --- Helper methods ---
     def _load_backend(self, backend_type: TrainingBackendEnum) -> TrainingBackend: