-
Notifications
You must be signed in to change notification settings - Fork 13
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Robert Luttrell <[email protected]>
- Loading branch information
Showing
2 changed files
with
48 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,11 +1,15 @@ | ||
# 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 | ||
|
||
from typing import Optional | ||
|
||
__all__ = ("PosixSessionUser", "SessionUser") | ||
|
@@ -38,7 +42,48 @@ 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 just a username for a local user, a domain user's UPN, or a domain user in down-level format. | ||
ex: localUser, [email protected], 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 format. | ||
ex: localGroup, domain\\domainGroup | ||
""" | ||
|
||
@staticmethod | ||
def upn_to_down_level(upn): | ||
return win32security.TranslateName( | ||
upn, win32api.NameUserPrincipal, win32api.NameSamCompatible | ||
) | ||
|
||
def __init__(self, user: str, *, group: str) -> None: | ||
""" | ||
Arguments: | ||
user (str): The user | ||
group (str): The group | ||
""" | ||
if not is_windows(): | ||
raise RuntimeError("Only available on Windows systems.") | ||
|
||
self.group = group | ||
|
||
try: | ||
self.user = self.upn_to_down_level(user) | ||
except Exception: | ||
raise RuntimeError("Unable to convert user {user} to down-level logon format.") |