feat: Add Windows session user

requirements-testing.txt

@@ -7,3 +7,4 @@ black >= 23.7, == 23.*
 ruff >= 0.0.286, == 0.0.*
 mypy >= 1.5.1, == 1.5.*
 psutil >= 5.9.5, == 5.9.*
+pywin32 == 306; platform_system == "Windows"

src/openjd/sessions/_session_user.py

@@ -1,14 +1,20 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 
 import os
-from ._os_checker import is_posix
+from ._os_checker import is_posix, is_windows
 
 if is_posix():
     import grp
 
+if is_windows():
+    import win32api
+    import win32security
+    import win32net
+    import win32netcon
+
 from typing import Optional
 
-__all__ = ("PosixSessionUser", "SessionUser")
+__all__ = ("PosixSessionUser", "SessionUser", "WindowsSessionUser")
 
 
 class SessionUser:
@@ -38,7 +44,63 @@ def __init__(self, user: str, *, group: Optional[str] = None) -> None:
             group (Optional[str]): The group. Defaults to the name of this
                 process' effective group.
         """
-        if os.name != "posix":
+        if not is_posix():
             raise RuntimeError("Only available on posix systems.")
         self.user = user
         self.group = group if group else grp.getgrgid(os.getegid()).gr_name  # type: ignore
+
+
+class WindowsSessionUser(SessionUser):
+    __slots__ = ("user", "group")
+    """Specific os-user identity to run a Session as under Windows."""
+
+    user: str
+    """
+    User name of the identity to run the Session's subprocesses under.
+    This can be either a plain username for a local user or a domain username in down-level logon form
+    ex: localUser, domain\\domainUser
+    """
+
+    group: str
+    """
+    Group name of the identity to run the Session's subprocesses under.
+    This can be just a group name for a local group, or a domain group in down-level logon form.
+    ex: localGroup, domain\\domainGroup
+    """
+
+    password: str
+    """
+    Password of the identity to run the Session's subprocess under.
+    """
+
+    @staticmethod
+    def is_domain_joined() -> bool:
+        """
+        Returns true if the machine is joined to a domain, otherwise False.
+        """
+        _, join_status = win32net.NetGetJoinInformation()
+        return join_status != win32netcon.NetSetupUnjoined
+
+    def __init__(self, user: str, password: str, *, group: str) -> None:
+        """
+        Arguments:
+            user (str): User name of the identity to run the Session's subprocesses under.
+                        This can be either a plain username for a local user or a domain username in down-level logon form
+                        ex: localUser, domain\\domainUser, [email protected]
+            group (str): Group name of the identity to run the Session's subprocesses under.
+                         This can be just a group name for a local group, or a domain group in down-level format.
+                         ex: localGroup, domain\\domainGroup
+            password (str): Password of the identity to run the Session's subprocess under.
+        """
+        if not is_windows():
+            raise RuntimeError("Only available on Windows systems.")
+
+        self.group = group
+        self.password = password
+
+        if "@" in user and self.is_domain_joined():
+            user = win32security.TranslateName(
+                user, win32api.NameUserPrincipal, win32api.NameSamCompatible
+            )
+
+        self.user = user

test/openjd/sessions/test_session_user.py

@@ -0,0 +1,18 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+from openjd.sessions._session_user import WindowsSessionUser
+from openjd.sessions._os_checker import is_windows
+
+import pytest
+
+
+@pytest.mark.skipif(not is_windows(), reason="Windows-specific tests")
+class TestWindowsSessionUser:
+    @pytest.mark.parametrize(
+        "user",
+        ["userA", "domain\\userA"],
+    )
+    def test_user_not_converted(self, user):
+        windows_session_user = WindowsSessionUser(user, group="test_group")
+
+        assert windows_session_user.user == user