From 5060981f30e6d1d4c980c89751610dcdffca0dbf Mon Sep 17 00:00:00 2001
From: Enno Gotthold <egotthold@suse.de>
Date: Mon, 30 Sep 2024 10:50:51 +0200
Subject: [PATCH] XML-RPC: Prevent privilege escalation from none to admin

This patch fixes
- GHSA-m26c-fcgh-cp6h
- CVE-2024-47533

It fixes two issues:

1. The encoding keyword argument isn't present when reading in binary mode. We now read in text mode.
2. The login function shouldn't compare the special error value -1 from "utils.get_shared_secret()" to the passed password.

Backported-by: fuowang <wangshuo@kylinos.cn>
---
 cobbler/cobblerd.py                   |  9 +++----
 cobbler/remote.py                     |  2 ++
 cobbler/utils.py                      |  9 +++----
 tests/utils_test.py                   | 27 ++++++++++++++++---
 tests/xmlrpcapi/miscellaneous_test.py | 39 +++++++++++++++++++++++++++
 5 files changed, 72 insertions(+), 14 deletions(-)

diff --git a/cobbler/cobblerd.py b/cobbler/cobblerd.py
index 9927160..a973ac8 100644
--- a/cobbler/cobblerd.py
+++ b/cobbler/cobblerd.py
@@ -53,13 +53,10 @@ def regen_ss_file():
     :return: 1 if this was successful.
     """
     ssfile = "/var/lib/cobbler/web.ss"
-    fd = open("/dev/urandom", 'rb')
-    data = fd.read(512)
-    fd.close()
+    data = os.urandom(512)
 
-    fd = os.open(ssfile, os.O_CREAT | os.O_RDWR, 0o660)
-    os.write(fd, binascii.hexlify(data))
-    os.close(fd)
+    with open(ssfile, 'w', 0o660, encoding="UTF-8") as fd:
+        fd.write(str(binascii.hexlify(data)))
     os.chmod(ssfile, 0o660)
 
     http_user = "apache"
diff --git a/cobbler/remote.py b/cobbler/remote.py
index 9ac0f8d..0756c1f 100644
--- a/cobbler/remote.py
+++ b/cobbler/remote.py
@@ -3219,6 +3219,8 @@ class CobblerXMLRPCInterface(object):
         """
         # if shared secret access is requested, don't bother hitting the auth plugin
         if login_user == "":
+            if self.shared_secret == -1:
+                raise ValueError("login failed(<DIRECT>)")
             if login_password == self.shared_secret:
                 return self.__make_token("<DIRECT>")
             else:
diff --git a/cobbler/utils.py b/cobbler/utils.py
index fc1939b..cfed465 100644
--- a/cobbler/utils.py
+++ b/cobbler/utils.py
@@ -2288,13 +2288,12 @@ def get_shared_secret():
     :return: The Cobbler secret which enables full access to Cobbler.
     :rtype: str
     """
-
     try:
-        with open("/var/lib/cobbler/web.ss", 'rb', encoding='utf-8') as fd:
-            data = fd.read()
-    except:
+        with open("/var/lib/cobbler/web.ss", "r", encoding="UTF-8") as web_secret_fd:
+            data = web_secret_fd.read()
+    except Exception:
         return -1
-    return str(data).strip()
+    return data
 
 
 def local_get_cobbler_api_url():
diff --git a/tests/utils_test.py b/tests/utils_test.py
index 5df3317..29edc45 100644
--- a/tests/utils_test.py
+++ b/tests/utils_test.py
@@ -1,8 +1,10 @@
+import binascii
 import os
 import re
 import shutil
 import time
 from pathlib import Path
+from typing import Any, TYPE_CHECKING
 
 import pytest
 from netaddr.ip import IPAddress
@@ -17,6 +19,9 @@ from cobbler.items.system import System
 from cobbler.cobbler_collections.manager import CollectionManager
 from tests.conftest import does_not_raise
 
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
 
 def test_pretty_hex():
     # Arrange
@@ -1037,15 +1042,31 @@ def test_load_signatures():
     assert old_cache != utils.SIGNATURE_CACHE
 
 
-def test_get_shared_secret():
+@pytest.mark.parametrize("web_ss_exists", [True, False])
+def test_get_shared_secret(mocker: "MockerFixture", web_ss_exists: bool):
     # Arrange
-    # TODO: Test the case where the file is there.
+    open_mock = mocker.mock_open()
+    random_data = binascii.hexlify(os.urandom(512)).decode()
+    mock_web_ss = mocker.mock_open(read_data=random_data)
+
+    def mock_open(*args: Any, **kwargs: Any):
+        if not web_ss_exists:
+            open_mock.side_effect = FileNotFoundError
+            return open_mock(*args, **kwargs)
+        if args[0] == "/var/lib/cobbler/web.ss":
+            return mock_web_ss(*args, **kwargs)
+        return open_mock(*args, **kwargs)
+
+    mocker.patch("builtins.open", mock_open)
 
     # Act
     result = utils.get_shared_secret()
 
     # Assert
-    assert result == -1
+    if web_ss_exists:
+        assert result == random_data
+    else:
+        assert result == -1
 
 
 def test_local_get_cobbler_api_url():
diff --git a/tests/xmlrpcapi/miscellaneous_test.py b/tests/xmlrpcapi/miscellaneous_test.py
index 1cdf249..3c0c917 100644
--- a/tests/xmlrpcapi/miscellaneous_test.py
+++ b/tests/xmlrpcapi/miscellaneous_test.py
@@ -1,11 +1,15 @@
 import json
 import os
 import time
+from typing import Any
 
 import pytest
 
+from cobbler.remote import CobblerXMLRPCInterface
 from cobbler.utils import get_shared_secret
 
+from tests.conftest import does_not_raise
+
 
 @pytest.mark.usefixtures("cobbler_xmlrpc_base")
 class TestMiscellaneous:
@@ -355,6 +359,41 @@ class TestMiscellaneous:
         # Assert
         assert not result
 
+    @pytest.mark.parametrize(
+        "input_username,input_password,expected_result,expected_exception,web_ss_exists",
+        [
+            ("cobbler", "cobbler", True, does_not_raise(), True),
+            ("cobbler", "incorrect-password", True, pytest.raises(ValueError), True),
+            ("", "doesnt-matter", True, pytest.raises(ValueError), True),
+            ("", "my-random-web-ss", True, does_not_raise(), True),
+            ("", "my-random-web-ss", True, pytest.raises(ValueError), False),
+        ],
+    )
+    def test_login(
+            self,
+            remote: CobblerXMLRPCInterface,
+            input_username: str,
+            input_password: str,
+            expected_result: Any,
+            expected_exception: Any,
+            web_ss_exists: bool
+    ):
+        """
+        Assert that the login is working successfully with correct and incorrect credentials.
+        """
+        # Arrange
+        if web_ss_exists:
+            remote.shared_secret = "my-random-web-ss"
+        else:
+            remote.shared_secret = -1
+
+        # Act
+        with expected_exception:
+            token = remote.login(input_username, input_password)
+
+            # Assert
+            assert remote.token_check(token) == expected_result
+
     def test_logout(self, remote):
         # Arrange
         shared_secret = get_shared_secret()
-- 
2.27.0

