From 40db387f19a6b1bedd0887a37bddfdfce6cad8a9 Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:39:07 -0700 Subject: [PATCH 1/2] fix: take the named pipe user SID from the process token NamedPipeHelper.create_security_attributes() resolved the DACL's user SID with LookupAccountName("", getpass.getuser()). getpass.getuser() only reads the LOGNAME, USER, LNAME and USERNAME environment variables, and the name lookup goes through LSA, which can fail with ERROR_NONE_MAPPED (1332). A Windows CI host running as SYSTEM hit 1332 on both the server path (create_named_pipe_server) and the client path (establish_named_pipe_connection), and the 3-attempt retry from #262 ran out. - Read the SID from the process token: OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY) and GetTokenInformation(token, TokenUser)[0], closing the token in a finally. No name lookup, so 1332 cannot occur. The #262 retry and the getpass import are removed. - The client no longer builds a DACL. CreateFile ignores the security descriptor when opening an existing pipe and reads only bInheritHandle. It passes SECURITY_ATTRIBUTES with SECURITY_DESCRIPTOR = None and bInheritHandle = True. pywin32's SECURITY_ATTRIBUTES() defaults bInheritHandle to TRUE (PySECURITY_ATTRIBUTES.cpp), so the client handle stays inheritable, as before. Passing None would have made it non-inheritable. - The server DACL is unchanged: GENERIC_READ|GENERIC_WRITE allowed to the user SID, then GENERIC_ALL denied to S-1-5-2 (NETWORK). Tests: - Unit: the token SID ends up in the DACL; with the env user name set to an unmapped name, LookupAccountName is not called; the token is closed when GetTokenInformation raises; the client passes no descriptor, an inheritable handle and OPEN_EXISTING, and does not call create_security_attributes. The three LookupAccountName/retry tests pinned the removed behaviour and are replaced. - Integ: test_pipe_security_does_not_depend_on_user_name runs a real pipe round-trip with LOGNAME, USER, LNAME and USERNAME set to no-such-user-1332, once for server and client and once for the client only, plus an ambient-environment control. It asserts the precondition that LookupAccountName("", "no-such-user-1332") raises 1332, and reads the DACL back with GetSecurityInfo. The old code fails this test deterministically. - CI: code_quality.yml runs windows-latest x Python 3.9-3.14 through OpenJobDescription/.github reusable_python_build.yml, whose last step (line 72) is `hatch run test`, which is `pytest ... test -vv`. That collects test/openjd/adaptor_runtime/integ/_named_pipe/. No marker gates the new test. Ran on macOS (Python 3.13, hatch 1.14.1): - hatch run lint, before and after: ruff "All checks passed!", black "146 files would be left unchanged", mypy "no issues found in 146 source files". - hatch run mypy --platform win32 src test, before and after: no issues in 146 source files. - hatch run test: before (mainline d9044ea) 482 passed, 26 skipped, coverage 93.39%. After: 482 passed, 29 skipped, coverage 93.39%. The 3 new skips are the 3 new integ cases. The Windows-only unit module is one module-level skip before and after. - Windows tests skip on macOS. To run them anyway, a fake pywin32 plugin (not committed) models the APIs these tests touch, using the pywin32 C sources. Fidelity control: mainline code with mainline tests passes on it (11 passed). Branch: 15 passed, 1 skipped (the GITHUB_ACTIONS-only different-user test). Mutation check, EXECUTED on the fake (sha256 restore, __pycache__ cleared between mutants), all caught: - M1 restore LookupAccountName(getpass.getuser()) plus retry: fails test_create_security_attributes_uses_process_token_user, ..._does_not_look_up_user_name, ..._closes_token_when_query_fails and integ [unmapped-user-name-for-server-and-client]. - M2 restore the client DACL: fails test_establish_named_pipe_connection_builds_no_dacl. - M3 original code (M1+M2): also fails integ [unmapped-user-name-for-client]. - M4 no finally, M5 no close, M6 client not inheritable, M7 client None, M8 client keeps blank descriptor, M9 allow Everyone, M10 drop network deny: each caught. M4-M8 by the token-close or client unit test, M9 and M10 also by all three integ cases. ARGUED for real Windows, not run: with the old code, LookupAccountName raises 1332 on every attempt (the test asserts this as a precondition), so both unmapped integ cases fail. M2 alone is equivalent at the OS level, because CreateFile ignores the descriptor. Only the unit test catches it. Auditors (three, independent; each returned PASS): - Pin/mutation: 27 of their own mutants, 26 killed. The survivor removes the bInheritHandle = True line, which is equivalent because pywin32 defaults it to TRUE. Kept the explicit line for the reader. The CloseHandle-call assertion rejects an equivalent token.Close(). Kept, since it matches the repo's mock style. - Gap analysis: 9 mutants, all killed. Pre-existing: test_fail_to_connect_to_named_pipe_with_different_user has its assert inside pytest.raises, so the assert never runs. Not in this diff, and tightening it to winerror 5 would gate CI on a claim nobody has run. Recorded as a follow-up. _secure_open.py has the same LookupAccountName pattern; follow-up. - Craft/cross-platform: lint, win32 mypy and collection green; pywin32 API shapes checked against source. Fixed: the client-only case now sets the unmapped name inside the try, so a failed precondition still closes the server handle. Declined: cross-checking the SID with an LSA name lookup. It would bring back the lookup this fix removes. The test helper hard-codes TokenUser independently, and a TokenOwner mutant is killed. Follow-up, out of scope: openjd-sessions-for-python _get_process_user_sid() uses the same name-lookup pattern. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- .../named_pipe/named_pipe_helper.py | 41 +++--- .../_named_pipe/test_named_pipe_helper.py | 137 +++++++++++++++++ .../unit/named_pipe/test_named_pipe_helper.py | 139 +++++++++++++----- 3 files changed, 260 insertions(+), 57 deletions(-) diff --git a/src/openjd/adaptor_runtime_client/named_pipe/named_pipe_helper.py b/src/openjd/adaptor_runtime_client/named_pipe/named_pipe_helper.py index 82b664c1..9416d15c 100644 --- a/src/openjd/adaptor_runtime_client/named_pipe/named_pipe_helper.py +++ b/src/openjd/adaptor_runtime_client/named_pipe/named_pipe_helper.py @@ -4,11 +4,11 @@ from concurrent.futures import ThreadPoolExecutor, TimeoutError from random import randint +import win32api import win32file import pywintypes import win32security import win32con -import getpass import winerror import time import win32pipe @@ -116,7 +116,7 @@ class NamedPipeHelper: """ @staticmethod - def create_security_attributes(): + def create_security_attributes() -> win32security.SECURITY_ATTRIBUTES: """ Creates and returns security attributes for a named pipe, allowing access only to the current user and denying network access. @@ -125,26 +125,14 @@ def create_security_attributes(): win32security.SECURITY_ATTRIBUTES: A SECURITY_ATTRIBUTES object configured with the custom security descriptor. """ - # Get the username of the current user - username = getpass.getuser() - - # Retry with exponential backoff to handle ERROR_NONE_MAPPED (1332) which occurs - # when the LSA service hasn't finished initializing on freshly started EC2 instances. - _LOOKUP_MAX_RETRIES = 3 - for attempt in range(_LOOKUP_MAX_RETRIES): - try: - user_sid, _, _ = win32security.LookupAccountName( - "", # systemName: The name of the system or server where the account resides. - # Search for the account on the local computer. - # If Domain/User Format is used here, it will fetch the Name from the AD. - username, - ) - break - except pywintypes.error as e: - if e.winerror == 1332 and attempt < _LOOKUP_MAX_RETRIES - 1: - time.sleep(2**attempt) - continue - raise + # Take the SID from the process token, not from a user name. A name lookup goes through + # LSA and can fail with ERROR_NONE_MAPPED (1332), and getpass.getuser() only reads + # environment variables, which need not name the process's account. + token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY) + try: + user_sid = win32security.GetTokenInformation(token, win32security.TokenUser)[0] + finally: + win32api.CloseHandle(token) # Users who log on across a network. "S-1-5-2" is a group identifier added to the token of a process # when it was logged on across a network. @@ -320,6 +308,13 @@ def establish_named_pipe_connection(pipe_name: str, timeout_in_seconds: float) - or due to other errors. """ + # CreateFile ignores the security descriptor when opening an existing pipe and reads only + # bInheritHandle, so build no DACL here. Inheritable matches pywin32's SECURITY_ATTRIBUTES + # default, which the server pipe handle also uses. + security_attributes = win32security.SECURITY_ATTRIBUTES() + security_attributes.SECURITY_DESCRIPTOR = None + security_attributes.bInheritHandle = True + start_time = time.time() # Wait for the server pipe to become available. handle = None @@ -330,7 +325,7 @@ def establish_named_pipe_connection(pipe_name: str, timeout_in_seconds: float) - # Give the read / write permission win32file.GENERIC_READ | win32file.GENERIC_WRITE, 0, # Disable the sharing Mode - NamedPipeHelper.create_security_attributes(), + security_attributes, win32file.OPEN_EXISTING, # Open existing pipe 0, # No Additional flags None, # A valid handle to a template file, This parameter is ignored when opening an existing pipe. diff --git a/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py b/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py index 84dcba9f..137a59ed 100644 --- a/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py +++ b/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py @@ -2,17 +2,23 @@ import time from openjd.adaptor_runtime._osname import OSName +import getpass import json import os import pytest import threading +import uuid +from typing import Any, List, Optional, Tuple if OSName.is_windows(): + import ntsecuritycon import pywintypes + import win32con import win32file import win32pipe import win32security import win32api + import winerror from openjd.adaptor_runtime_client.named_pipe.named_pipe_helper import NamedPipeHelper else: # Cannot put this on the top of this file or mypy will complain @@ -21,6 +27,12 @@ PIPE_NAME = r"\\.\pipe\TestPipe" TIMEOUT_SECONDS = 5 +# getpass.getuser() returns the first of these that is set. No account has this name, so +# resolving it with LookupAccountName fails with ERROR_NONE_MAPPED (1332). +USER_NAME_ENV_VARS = ("LOGNAME", "USER", "LNAME", "USERNAME") +UNMAPPED_USER_NAME = "no-such-user-1332" +NETWORK_SID = "S-1-5-2" + def pipe_server(pipe_name, message_to_send, return_message): """ @@ -52,6 +64,64 @@ def start_pipe_server(): server_thread.join() +def set_unmapped_user_name(monkeypatch: pytest.MonkeyPatch) -> None: + """Makes getpass.getuser() return a name that LookupAccountName cannot resolve.""" + for name in USER_NAME_ENV_VARS: + monkeypatch.setenv(name, UNMAPPED_USER_NAME) + assert getpass.getuser() == UNMAPPED_USER_NAME + # Precondition: if this name resolved, a name-based SID lookup would pass too and the + # tests using it would pin nothing. + with pytest.raises(pywintypes.error) as excinfo: + win32security.LookupAccountName("", UNMAPPED_USER_NAME) + assert excinfo.value.winerror == winerror.ERROR_NONE_MAPPED + + +def process_token_user_sid() -> str: + token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY) + try: + sid = win32security.GetTokenInformation(token, win32security.TokenUser)[0] + finally: + win32api.CloseHandle(token) + return win32security.ConvertSidToStringSid(sid) + + +def as_file_rights(mask: int) -> int: + """Maps generic rights to file rights, since Windows may store an ACE mask in either form.""" + mask &= 0xFFFFFFFF + for generic, specific in ( + (win32con.GENERIC_READ & 0xFFFFFFFF, ntsecuritycon.FILE_GENERIC_READ), + (win32con.GENERIC_WRITE, ntsecuritycon.FILE_GENERIC_WRITE), + (win32con.GENERIC_ALL, ntsecuritycon.FILE_ALL_ACCESS), + ): + if mask & generic: + mask = (mask & ~generic) | specific + return mask + + +def read_pipe_dacl(pipe_handle: Any) -> List[Tuple[int, int, str]]: + """Returns (ACE type, access mask as file rights, SID string) for each ACE on the pipe.""" + security_descriptor = win32security.GetSecurityInfo( + pipe_handle, win32security.SE_KERNEL_OBJECT, win32security.DACL_SECURITY_INFORMATION + ) + dacl = security_descriptor.GetSecurityDescriptorDacl() + entries = [] + for index in range(dacl.GetAceCount()): + (ace_type, _ace_flags), mask, sid = dacl.GetAce(index) + entries.append((ace_type, as_file_rights(mask), win32security.ConvertSidToStringSid(sid))) + return entries + + +def serve_one_request( + server_handle: Any, response: str, received: List[str], errors: List[Exception] +) -> None: + try: + win32pipe.ConnectNamedPipe(server_handle, None) + received.append(NamedPipeHelper.read_from_pipe(server_handle)) + NamedPipeHelper.write_to_pipe(server_handle, response) + except Exception as e: + errors.append(e) + + @pytest.mark.skipif(not OSName.is_windows(), reason="NamedPipe is only implemented in Windows.") class TestNamedPipeHelper: def test_named_pipe_communication(self, start_pipe_server): @@ -129,3 +199,70 @@ def test_fail_to_connect_to_named_pipe_with_different_user( # Send a message to unblock the I/O NamedPipeHelper.send_named_pipe_request(PIPE_NAME, TIMEOUT_SECONDS, **message_to_send) + + @pytest.mark.parametrize( + "unmapped_user_name_for", + [ + pytest.param(None, id="ambient-user-name"), + pytest.param("server-and-client", id="unmapped-user-name-for-server-and-client"), + pytest.param("client", id="unmapped-user-name-for-client"), + ], + ) + def test_pipe_security_does_not_depend_on_user_name( + self, monkeypatch: pytest.MonkeyPatch, unmapped_user_name_for: Optional[str] + ) -> None: + """ + The pipe's DACL comes from the process token, so a getpass.getuser() name that maps to + no account (ERROR_NONE_MAPPED, 1332) breaks neither the server nor the client. + The ambient-user-name case is the control: the normal environment must still work. + """ + # GIVEN + pipe_name = rf"\\.\pipe\openjd_token_sid_{os.getpid()}_{uuid.uuid4().hex}" + request_body = {"message": "Hello from client"} + server_response = '{"Response": "Hello from server"}' + received: List[str] = [] + errors: List[Exception] = [] + if unmapped_user_name_for == "server-and-client": + set_unmapped_user_name(monkeypatch) + server_handle = NamedPipeHelper.create_named_pipe_server(pipe_name, TIMEOUT_SECONDS) + assert server_handle is not None + # A daemon thread, so a failed client cannot hang the run in ConnectNamedPipe. + server_thread = threading.Thread( + target=serve_one_request, + args=(server_handle, server_response, received, errors), + daemon=True, + ) + + try: + if unmapped_user_name_for == "client": + set_unmapped_user_name(monkeypatch) + dacl = read_pipe_dacl(server_handle) + server_thread.start() + + # WHEN + response = NamedPipeHelper.send_named_pipe_request( + pipe_name, TIMEOUT_SECONDS, "POST", "/token-sid", json_body=request_body + ) + server_thread.join(TIMEOUT_SECONDS) + finally: + win32file.CloseHandle(server_handle) + + # THEN + assert not server_thread.is_alive() + assert errors == [] + assert [json.loads(message) for message in received] == [ + {"method": "POST", "path": "/token-sid", "body": json.dumps(request_body)} + ] + assert response == json.loads(server_response) + assert dacl == [ + ( + win32security.ACCESS_ALLOWED_ACE_TYPE, + as_file_rights(win32con.GENERIC_READ | win32con.GENERIC_WRITE), + process_token_user_sid(), + ), + ( + win32security.ACCESS_DENIED_ACE_TYPE, + as_file_rights(win32con.GENERIC_ALL), + NETWORK_SID, + ), + ] diff --git a/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py b/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py index 0c8237d1..f69a5f69 100644 --- a/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py +++ b/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py @@ -1,19 +1,35 @@ # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. from openjd.adaptor_runtime._osname import OSName +from typing import Any, List, Tuple from unittest.mock import patch, MagicMock import pytest import os import time pywintypes = pytest.importorskip("pywintypes") +win32api = pytest.importorskip("win32api") +win32con = pytest.importorskip("win32con") win32pipe = pytest.importorskip("win32pipe") win32file = pytest.importorskip("win32file") +win32security = pytest.importorskip("win32security") winerror = pytest.importorskip("winerror") named_pipe_helper = pytest.importorskip( "openjd.adaptor_runtime_client.named_pipe.named_pipe_helper" ) +FAKE_USER_SID = "S-1-5-21-1-2-3-1001" + + +def dacl_entries(security_attributes: Any) -> List[Tuple[int, int, str]]: + """Returns (ACE type, access mask, SID string) for each ACE in the attributes' DACL.""" + dacl = security_attributes.SECURITY_DESCRIPTOR.GetSecurityDescriptorDacl() + entries = [] + for index in range(dacl.GetAceCount()): + (ace_type, _ace_flags), mask, sid = dacl.GetAce(index) + entries.append((ace_type, mask & 0xFFFFFFFF, win32security.ConvertSidToStringSid(sid))) + return entries + class MockReadFile: @staticmethod @@ -101,42 +117,97 @@ def test_failed_to_generate_pipe_name(self, mock_check_named_pipe_exists, mock_g ): named_pipe_helper.NamedPipeHelper.generate_pipe_name("AdaptorTest") - @patch("getpass.getuser", return_value="regularuser") - def test_create_security_attributes_uses_lookup(self, mock_getuser): - win32security = pytest.importorskip("win32security") - fake_sid = win32security.ConvertStringSidToSid("S-1-1-0") - with patch.object( - win32security, "LookupAccountName", return_value=(fake_sid, None, None) - ) as mock_lookup: - named_pipe_helper.NamedPipeHelper.create_security_attributes() - mock_lookup.assert_called_once_with("", "regularuser") - - @patch("time.sleep") - @patch("getpass.getuser", return_value="regularuser") - def test_create_security_attributes_retries_on_lsa_not_ready(self, mock_getuser, mock_sleep): - # ERROR_NONE_MAPPED (1332) is raised when LSA hasn't finished initializing on a - # fresh EC2 instance. The lookup should retry with exponential backoff and succeed. - win32security = pytest.importorskip("win32security") - fake_sid = win32security.ConvertStringSidToSid("S-1-1-0") - lsa_error = pywintypes.error( - 1332, "LookupAccountName", "No mapping between account names and security IDs was done." + def test_create_security_attributes_uses_process_token_user(self) -> None: + # GIVEN + token = MagicMock() + token_user = (win32security.ConvertStringSidToSid(FAKE_USER_SID), 0) + with ( + patch.object(win32security, "OpenProcessToken", return_value=token) as mock_open_token, + patch.object( + win32security, "GetTokenInformation", return_value=token_user + ) as mock_get_token_information, + patch.object(win32api, "CloseHandle") as mock_close_handle, + ): + # WHEN + security_attributes = named_pipe_helper.NamedPipeHelper.create_security_attributes() + + # THEN + mock_open_token.assert_called_once_with(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY) + mock_get_token_information.assert_called_once_with(token, win32security.TokenUser) + mock_close_handle.assert_called_once_with(token) + assert dacl_entries(security_attributes) == [ + ( + win32security.ACCESS_ALLOWED_ACE_TYPE, + (win32con.GENERIC_READ | win32con.GENERIC_WRITE) & 0xFFFFFFFF, + FAKE_USER_SID, + ), + (win32security.ACCESS_DENIED_ACE_TYPE, win32con.GENERIC_ALL, "S-1-5-2"), + ] + + def test_create_security_attributes_does_not_look_up_user_name( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # GIVEN + for name in ("LOGNAME", "USER", "LNAME", "USERNAME"): + monkeypatch.setenv(name, "no-such-user-1332") + none_mapped = pywintypes.error( + winerror.ERROR_NONE_MAPPED, + "LookupAccountName", + "No mapping between account names and security IDs was done.", ) + token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY) + try: + expected_sid = win32security.GetTokenInformation(token, win32security.TokenUser)[0] + finally: + win32api.CloseHandle(token) + with patch.object( - win32security, "LookupAccountName", side_effect=[lsa_error, (fake_sid, None, None)] - ) as mock_lookup: - named_pipe_helper.NamedPipeHelper.create_security_attributes() - assert mock_lookup.call_count == 2 - mock_sleep.assert_called_once_with(1) # 2**0 = 1s after first failure - - @patch("time.sleep") - @patch("getpass.getuser", return_value="regularuser") - def test_create_security_attributes_raises_after_max_retries(self, mock_getuser, mock_sleep): - # After exhausting all retries, the original error should be re-raised. - win32security = pytest.importorskip("win32security") - lsa_error = pywintypes.error( - 1332, "LookupAccountName", "No mapping between account names and security IDs was done." + win32security, "LookupAccountName", side_effect=none_mapped + ) as mock_lookup_account_name: + # WHEN + security_attributes = named_pipe_helper.NamedPipeHelper.create_security_attributes() + + # THEN + mock_lookup_account_name.assert_not_called() + allowed_sid = dacl_entries(security_attributes)[0][2] + assert allowed_sid == win32security.ConvertSidToStringSid(expected_sid) + + def test_create_security_attributes_closes_token_when_query_fails(self) -> None: + # GIVEN + token = MagicMock() + query_error = pywintypes.error( + winerror.ERROR_ACCESS_DENIED, "GetTokenInformation", "Access is denied." ) - with patch.object(win32security, "LookupAccountName", side_effect=lsa_error): + with ( + patch.object(win32security, "OpenProcessToken", return_value=token), + patch.object(win32security, "GetTokenInformation", side_effect=query_error), + patch.object(win32api, "CloseHandle") as mock_close_handle, + ): + # WHEN with pytest.raises(pywintypes.error) as exc_info: named_pipe_helper.NamedPipeHelper.create_security_attributes() - assert exc_info.value.winerror == 1332 + + # THEN + assert exc_info.value is query_error + mock_close_handle.assert_called_once_with(token) + + @patch.object(win32pipe, "SetNamedPipeHandleState") + @patch.object(win32file, "CreateFile") + @patch.object(named_pipe_helper.NamedPipeHelper, "create_security_attributes") + def test_establish_named_pipe_connection_builds_no_dacl( + self, + mock_create_security_attributes: MagicMock, + mock_create_file: MagicMock, + mock_set_named_pipe_handle_state: MagicMock, + ) -> None: + # WHEN + handle = named_pipe_helper.NamedPipeHelper.establish_named_pipe_connection("fakepipe", 1.0) + + # THEN + assert handle is mock_create_file.return_value + mock_create_security_attributes.assert_not_called() + # Opening an existing pipe ignores the security descriptor; inheritability is unchanged. + security_attributes = mock_create_file.call_args.args[3] + assert security_attributes.SECURITY_DESCRIPTOR is None + assert security_attributes.bInheritHandle == 1 + assert mock_create_file.call_args.args[4] == win32file.OPEN_EXISTING From 566de5c34cc7d42a3f16b384b10ea3fe8384a8ba Mon Sep 17 00:00:00 2001 From: David Leong <116610336+leongdl@users.noreply.github.com> Date: Mon, 2 Nov 2026 21:02:58 -0700 Subject: [PATCH 2/2] test: expect the deny-first ACE order on the named pipe DACL Windows CI showed the pipe DACL reads back with the NETWORK deny ACE before the user allow ACE, both in the in-memory security attributes and on the created pipe. That is canonical order, and the order that makes the deny apply to the user's own network logons. The new tests asserted allow-first and failed on all six Windows jobs; production code was not involved, since mainline builds the same DACL. Assert the canonical order exactly, and find the allowed SID by ACE type rather than by index. Signed-off-by: David Leong <116610336+leongdl@users.noreply.github.com> --- .../integ/_named_pipe/test_named_pipe_helper.py | 11 ++++++----- .../unit/named_pipe/test_named_pipe_helper.py | 11 ++++++++--- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py b/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py index 137a59ed..7a9a04cf 100644 --- a/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py +++ b/test/openjd/adaptor_runtime/integ/_named_pipe/test_named_pipe_helper.py @@ -254,15 +254,16 @@ def test_pipe_security_does_not_depend_on_user_name( {"method": "POST", "path": "/token-sid", "body": json.dumps(request_body)} ] assert response == json.loads(server_response) + # Canonical order: the NETWORK deny comes first, so it applies to the user's own network logons. assert dacl == [ - ( - win32security.ACCESS_ALLOWED_ACE_TYPE, - as_file_rights(win32con.GENERIC_READ | win32con.GENERIC_WRITE), - process_token_user_sid(), - ), ( win32security.ACCESS_DENIED_ACE_TYPE, as_file_rights(win32con.GENERIC_ALL), NETWORK_SID, ), + ( + win32security.ACCESS_ALLOWED_ACE_TYPE, + as_file_rights(win32con.GENERIC_READ | win32con.GENERIC_WRITE), + process_token_user_sid(), + ), ] diff --git a/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py b/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py index f69a5f69..a5c86f70 100644 --- a/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py +++ b/test/openjd/adaptor_runtime/unit/named_pipe/test_named_pipe_helper.py @@ -135,13 +135,14 @@ def test_create_security_attributes_uses_process_token_user(self) -> None: mock_open_token.assert_called_once_with(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY) mock_get_token_information.assert_called_once_with(token, win32security.TokenUser) mock_close_handle.assert_called_once_with(token) + # Canonical order: the NETWORK deny comes first, so it applies to the user's own network logons. assert dacl_entries(security_attributes) == [ + (win32security.ACCESS_DENIED_ACE_TYPE, win32con.GENERIC_ALL, "S-1-5-2"), ( win32security.ACCESS_ALLOWED_ACE_TYPE, (win32con.GENERIC_READ | win32con.GENERIC_WRITE) & 0xFFFFFFFF, FAKE_USER_SID, ), - (win32security.ACCESS_DENIED_ACE_TYPE, win32con.GENERIC_ALL, "S-1-5-2"), ] def test_create_security_attributes_does_not_look_up_user_name( @@ -169,8 +170,12 @@ def test_create_security_attributes_does_not_look_up_user_name( # THEN mock_lookup_account_name.assert_not_called() - allowed_sid = dacl_entries(security_attributes)[0][2] - assert allowed_sid == win32security.ConvertSidToStringSid(expected_sid) + allowed_sids = [ + sid + for ace_type, _mask, sid in dacl_entries(security_attributes) + if ace_type == win32security.ACCESS_ALLOWED_ACE_TYPE + ] + assert allowed_sids == [win32security.ConvertSidToStringSid(expected_sid)] def test_create_security_attributes_closes_token_when_query_fails(self) -> None: # GIVEN