🛡️ Sentinel: [SECURITY] Enforce secure URL validation and file permissions

.jules/sentinel.md

@@ -0,0 +1,6 @@
+# Sentinel's Journal
+
+## 2025-02-07 - Secure File Creation Pattern
+**Vulnerability:** Default `open()` creates files with user umask, which might allow group/world read access for sensitive files like `plan.json`.
+**Learning:** Python's `open()` mode `w` respects `umask` but doesn't enforce restrictive permissions by default. `os.chmod` after creation leaves a small race condition window.
+**Prevention:** Use `os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)` to create the file descriptor with correct permissions atomically, then wrap with `os.fdopen()`.

main.py

@@ -340,6 +340,12 @@
         if not hostname:
             return False
 
+        if parsed.username or parsed.password:
+            log.warning(
+                f"Skipping unsafe URL (credentials detected): {sanitize_for_log(url)}"
+            )
+            return False
+
         # Check for potentially malicious hostnames
         if hostname.lower() in ("localhost", "127.0.0.1", "::1"):
             log.warning(
@@ -514,7 +520,7 @@
             response = request_func()
             response.raise_for_status()
             return response
         except (httpx.HTTPError, httpx.TimeoutException) as e:
             if attempt == max_retries - 1:
                 if hasattr(e, "response") and e.response is not None:
                     log.debug(f"Response content: {sanitize_for_log(e.response.text)}")
@@ -1370,9 +1376,14 @@
         )
 
     if args.plan_json:
-        with open(args.plan_json, "w", encoding="utf-8") as f:
-            json.dump(plan, f, indent=2)
-        log.info("Plan written to %s", args.plan_json)
+        try:
+            # Securely create file with 600 permissions
+            fd = os.open(args.plan_json, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                json.dump(plan, f, indent=2)
+            log.info("Plan written to %s", args.plan_json)
+        except OSError as e:
+            log.error(f"Failed to write plan to {args.plan_json}: {e}")
 
     # Print Summary Table
     # Determine the width for the Profile ID column (min 25)

test_main.py

@@ -1,5 +1,6 @@
 import importlib
 import os
+import stat
 import sys
 from unittest.mock import MagicMock, call, patch
 
@@ -74,7 +75,7 @@
 
 
 # Case 3: push_rules updates data dictionary with pre-calculated batch keys correctly
 def test_push_rules_updates_data_with_batch_keys(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
     mock_post_form = MagicMock()
@@ -257,7 +258,7 @@
     assert m.check_api_access(mock_client, "valid_profile") is True
 
 
 def test_check_api_access_401(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
 
@@ -320,7 +321,7 @@
     assert "Profile Not Found" in str(mock_log.critical.call_args_list)
 
 
 def test_check_api_access_generic_http_error(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
 
@@ -510,3 +511,80 @@
     # Color codes (accessing instance Colors or m.Colors)
     assert m.Colors.CYAN in combined
     assert m.Colors.ENDC in combined
+
+
+# Case 14: validate_folder_url rejects URLs with credentials
+def test_validate_folder_url_rejects_credentials(monkeypatch):
+    m = reload_main_with_env(monkeypatch)
+    mock_log = MagicMock()
+    monkeypatch.setattr(m, "log", mock_log)
+
+    # Use a mock for socket.getaddrinfo so we don't hit network or fail DNS
+    # But validate_folder_url logic for credentials happens BEFORE DNS.
+    # So we don't strictly need to mock DNS if it returns early.
+    # However, if it falls through (fails logic), it hits DNS.
+    # To be safe and isolated, mock getaddrinfo.
+    with patch("socket.getaddrinfo") as mock_dns:
+        mock_dns.return_value = [(2, 1, 6, "", ("1.1.1.1", 443))]
+        # URL with credentials
+        url = "https://user:pass@example.com/list.json"
+        assert m.validate_folder_url(url) is False
+        assert mock_log.warning.called
+        # Check that warning message contains "credentials detected"
+        # call_args is (args, kwargs)
+        args, _ = mock_log.warning.call_args
+        assert "credentials detected" in args[0]
+
+
+# Case 15: plan.json is created with secure permissions
+def test_plan_json_secure_permissions(monkeypatch, tmp_path, capsys):
+    # Prepare environment
+    monkeypatch.setenv("TOKEN", "dummy")
+    m = reload_main_with_env(monkeypatch)
+
+    # Mock args
+    plan_file = tmp_path / "plan.json"
+    mock_args = MagicMock()
+    mock_args.plan_json = str(plan_file)
+    mock_args.dry_run = True
+    mock_args.profiles = "p1"
+    # Empty folder_url to avoid fetch logic
+    mock_args.folder_url = []
+    mock_args.no_delete = False
+
+    monkeypatch.setattr(m, "parse_args", lambda: mock_args)
+    m.TOKEN = "dummy"
+
+    # Mock warm_up_cache to avoid actual network
+    monkeypatch.setattr(m, "warm_up_cache", MagicMock())
+
+    # Mock sync_profile so it returns True and populates plan
+    def mock_sync(pid, urls, dry_run, no_delete, plan_accumulator=None):
+        if plan_accumulator is not None:
+            # Add a dummy plan entry
+            plan_accumulator.append({
+                "profile": pid,
+                "folders": [{"name": "F1", "rules": 10, "action": 0, "status": 1}]
+            })
+        return True
+
+    monkeypatch.setattr(m, "sync_profile", mock_sync)
+
+    # Run main, expect SystemExit(0) on success
+    with pytest.raises(SystemExit) as e:
+        m.main()
+    assert e.value.code == 0
+
+    # Verify file exists
+    assert plan_file.exists()
+
+    # Verify permissions on POSIX
+    if os.name != "nt":
+        mode = os.stat(plan_file).st_mode
+        # Check permissions are 600 (rw-------)
+        # S_IRWXG = 0o070 (group rwx)
+        # S_IRWXO = 0o007 (other rwx)
+        assert not (mode & stat.S_IRWXG), "Group should have no permissions"
+        assert not (mode & stat.S_IRWXO), "Others should have no permissions"
+        assert mode & stat.S_IRUSR, "Owner should have read"
+        assert mode & stat.S_IWUSR, "Owner should have write"