pypsrp - Python PowerShell Remoting Protocol Client library
pypsrp is a Python client for the PowerShell Remoting Protocol (PSRP) service. It allows you to execute PowerShell scripts inside the Python script with a target being remote or some other local process.
This library has a low level API designed to mirror the System.Management.Automation namespace. There are also some helper functions designed to make it easier to do one off scripts and copy/fetch files on the target PSSession.
The old pypsrp
namespace only supported WSMan based transports but the psrp
namespace supports the following:
- WSMan - targets only remote Windows hosts
- SSH - can target both remote Windows and non-Windows hosts
- Process - can spawn a new local pwsh or powershell process and run code in there
- Named Pipes - targets a named pipe, like the pwsh remoting pipe, and communicate over that
The WSMan connection supports the following authentication protocols out of the box:
- Negotiate (Default)
- Basic
- Certificate
- NTLM
- CredSSP
To support Kerberos the kerberos
extras package must be installed.
Requirements
See How to Install
for more details
- CPython 3.8+
- cryptography
- psrpcore
- pyspnego
- requests - for the
pypsrp
code - httpx - for the
psrp
code
Optional Requirements
The following Python libraries can be installed to add extra features that do not come with the base package:
- python-gssapi for Kerberos authentication on Linux
- pykrb5 for Kerberos authentication on Linux
- asyncssh for SSH connections
- psutil for Named Pipe connections
How to Install
To install pypsrp with all the basic features, run:
pip install pypsrp
Kerberos Authentication
While pypsrp supports Kerberos authentication, it isn't included by default for Linux hosts due to it's reliance on system packages to be present.
To install these packages, depending on your distribution, run one of the following script blocks.
For Debian/Ubuntu
# For Python 2
apt-get install gcc python-dev libkrb5-dev
# For Python 3
apt-get install gcc python3-dev libkrb5-dev
# To add NTLM to the GSSAPI SPNEGO auth run
apt-get install gss-ntlmssp
For RHEL/Centos
yum install gcc python-devel krb5-devel
# To add NTLM to the GSSAPI SPNEGO auth run
yum install gssntlmssp
For Fedora
dnf install gcc python-devel krb5-devel
# To add NTLM to the GSSAPI SPNEGO auth run
dnf install gssntlmssp
For Arch Linux
pacman -S gcc krb5
Once installed you can install the Python packages with
pip install pypsrp[kerberos]
Kerberos also needs to be configured to talk to the domain but that is outside the scope of this page.
SSH Connections
The SSH connection on psrp
requires the asyncssh
library to be installed.
pip install pypsrp[ssh]
Named Pipe Connections
The Named Pipe connection on psrp
requires the psutil
library to be installed.
pip install pypsrp[named_pipe]
How to Use
There are 3 main components that are in use within this library:
ConnectionInfo
: Defines the connection type and connection specific variablesRunspacePool
: The Runspace Pool contains a pool of pipelines that can be run on the remote targetPipeline
: The code to run inside the Runspace Pool
ConnectionInfo
These are the connection info types that are supported by pypsrp
Type | Sync | Asyncio | Mandatory Requirements | Optional Requirements |
---|---|---|---|---|
WSManInfo | Y | Y | N/A | pypsrp[kerberos] for Kerberos support |
ProcessInfo | Y | Y | N/A | N/A |
SSHInfo | N | Y | pypsrp[ssh] |
N/A |
NamedPipeInfo | N | Y | pypsrp[named_pipe] |
N/A |
The mandatory requirements are requirements that must be installed on top of what pypsrp
requires.
The optional requirements are requirements to utilise optional features that aren't available by default.
The connection info objects do not store the connections themselves, they just define how a Runspace Pool will connect to the target. This means they can be reused across multiple pools as needed.
The psrp.AsyncOutOfProcConnection
and psrp.SyncOutOfProcConnection
can also be used to define your own out of process connection type.
This is fairly advanced work as it would require an implementation on both the client and server side.
RunspacePool
The Runspace Pool is used to create the connection to the remote target and can host multiple pipelines that run code. A Runspace Pool comes in 2 varieties:
psrp.SyncRunspacePool
- uses a synchronous connectionpsrp.AsyncRunspacePool
- uses an asyncio based connection
Both of these types must be created with a ConnectionInfo
that describes how to connect to the remote PowerShell instance.
See the table in ConnectionInfo to see what connections are supported by a syncronous Runspace Pool and an asyncronous Runspace Pool.
import psrp
async def async_rp(conn: psrp.ConnectionInfo) -> None:
async with psrp.AsyncRunspacePool(conn) as rp:
...
def sync_rp(conn: psrp.ConnectionInfo) -> None:
with psrp.SyncRunspacePool(conn) as rp:
...
Both the sync and async Runspace Pool contain the same methods and functionality, the main difference is that most operations on the async pool are coroutines that need to be awaited.
Pipeline
A Pipeline is used to execute a command or script on the Runspace Pool it is associated with. There are 4 types of pipelines that can be used:
psrp.AsyncPowerShell
- runs a PowerShell command through asynciopsrp.SyncPowerShell
- runs a PowerShell command through synchronous codepsrp.AsyncCommandMetaPipeline
- gets command metadata on the Runspace Pool through asynciopsrp.SyncCommandMetaPipeline
- gets command metadata on the Runspace Pool through syncronous code
The PowerShell pipeline is the commonly used pipeline that can run PowerShell commands, statements, and/or scripts.
Examples
Running PowerShell script
import psrp
async def async_rp(conn: psrp.ConnectionInfo) -> None:
async with psrp.AsyncRunspacePool(conn) as rp:
ps = psrp.AsyncPowerShell(rp)
ps.add_script('echo "hi"')
output = await ps.invoke()
print(output)
def sync_rp(conn: psrp.ConnectionInfo) -> None:
with psrp.SyncRunspacePool(conn) as rp:
ps = psrp.SyncPowerShell(rp)
ps.add_script('echo "hi"')
output = ps.invoke()
print(output)
This will run a PowerShell script and print out the output from that script.
The output from invoke()
is a list of PowerShell objects that are output from the remote pipeline.
Run a PowerShell command
import psrp
async def async_rp(conn: psrp.ConnectionInfo) -> None:
async with psrp.AsyncRunspacePool(conn) as rp:
ps = psrp.AsyncPowerShell(rp)
ps.add_command("Get-Process").add_command("Select-Object").add_parameter("Property", "Name")
ps.add_statement()
ps.add_command("Get-Service").add_argument("audiosrc")
output = await ps.invoke()
print(output)
def sync_rp(conn: psrp.ConnectionInfo) -> None:
with psrp.SyncRunspacePool(conn) as rp:
ps = psrp.AsyncPowerShell(rp)
ps.add_command("Get-Process").add_command("Select-Object").add_parameter("Property", "Name")
ps.add_statement()
ps.add_command("Get-Service").add_argument("audiosrc")
output = ps.invoke()
print(output)
This will run the PowerShell command Get-Process | Select-Object -Property Name; Get-Service audiosrv
.
Each command in a statement are piped and parameters/arguments are added to the last command.
The statement will run as a separate line/statement in the script.
Copy a file to the remote host
import psrp
def copy_file(conn: psrp.ConnectionInfo) -> None:
psrp.copy_file(conn, "/tmp/test.txt", r"C:\temp\test.txt")
Copies a local file to the remote PowerShell session.
Note: There is no asyncio analogue for this operation due to a lack of asyncio file libraries in the stdlib.
Fetches a file from the remote host
import psrp
def fetch_file(conn: psrp.ConnectionInfo) -> None:
psrp.fetch_file(conn, r"C:\temp\test.txt", "/tmp/test.txt")
Fetches a remote file to the local filesystem.
Note: There is no asyncio analogue for this operation due to a lack of asyncio file libraries in the stdlib.
Run script with high level API
import psrp
async def async_invoke_ps(conn: psrp.ConnectionInfo, script: str) -> None:
out, streams, had_errors = await psrp.async_invoke_ps(script)
print(f"OUTPUT: {out}")
if had_errors:
errors = [str(e) for e in streams.error]
print(f"ERROR: {errors}")
async def invoke_ps(conn: psrp.ConnectionInfo, script: str) -> None:
out, streams, had_errors = psrp.invoke_ps(script)
print(f"OUTPUT: {out}")
if had_errors:
errors = [str(e) for e in streams.error]
print(f"ERROR: {errors}")
Uses the high level API to execute a PowerShell script and print out any errors that are returned.
Authenticating with Exchange Online
This shows you how to connect against Exchange Online with the Python MSAL library.
import hashlib
import msal
import psrp
from cryptography.hazmat.primitives.serialization import (
Encoding,
NoEncryption,
PrivateFormat,
)
from cryptography.hazmat.primitives.serialization.pkcs12 import (
load_key_and_certificates,
)
def get_msal_token(
organization: str,
client_id: str,
pfx_path: str,
pfx_password: str | None,
) -> str:
private_key, main_cert, add_certs = load_key_and_certificates(
pfx,
pfx_password.encode("utf-8") if pfx_password else None,
None
)
assert private_key is not None
assert main_cert is not None
key = private_key.private_bytes(
Encoding.PEM,
PrivateFormat.PKCS8,
NoEncryption(),
).decode()
cert_thumbprint = hashlib.sha1()
cert_thumbprint.update(main_cert.public_bytes(Encoding.DER))
app = msal.ConfidentialClientApplication(
authority=f"https://login.microsoftonline.com/{organization}",
client_id=client_id,
client_credential={
"private_key": key,
"thumbprint": cert_thumbprint.hexdigest().upper(),
},
)
result = app.acquire_token_for_client(scopes=[
"https://outlook.office365.com/.default"
])
if err := result.get("error", None):
msg = f"Failed to get MSAL token {err} - {result['error_description']}"
raise Exception(msg)
return f"{result['token_type']} {result['access_token']}"
def main() -> None:
tenant_id = "00000000-0000-0000-0000-000000000000"
# This is the ID of the Application Role to authenticate as
client_id = "00000000-0000-0000-0000-000000000000"
msal_token = get_msal_token(
"test.onmicrosoft.com",
client_id,
"exchange.pfx",
"cert-password"
)
conn_info = psrp.WSManInfo(
server="https://outlook.office365.com/PowerShell-LiveId?BasicAuthToOAuthConversion=true",
auth="basic",
username=f"OAuthUser@{tenant_id}",
password=msal_token,
configuration_name="Microsoft.Exchange",
)
with psrp.SyncRunspacePool(conn_info) = rp:
ps = psrp.SyncPowerShell(rp)
ps.add_command("Get-Mailbox")
print(ps.invoke())
Logging
This library takes advantage of the Python logging configuration and messages are logged to the following named loggers
psrp.*
- The loggers for code under thepsrp
namespacepypsrp.*
- The loggers for the code under the legacypypsrp
namespace
Note: DEBUG
contains a lot of information and will output all the messages
sent to and from the client. This can have the side effect of leaking sensitive
information and should only be used for debugging purposes.
Testing
Any changes are more than welcome in pull request form, you can run the current test suite with tox like so;
# make sure tox is installed
pip install tox
# run the tox suite
tox
# or run the test manually for the current Python environment
python -m pytest tests/tests_psrp -v --cov psrp --cov-report term-missing
A lot of the tests either simulate a remote Windows host but you can also run a lot of them against a real Windows host. To do this, set the following environment variables before running the tests;
PYPSRP_SERVER
: The hostname or IP of the remote host to test WSMan withPYPSRP_USERNAME
: The username to use with WSManPYPSRP_PASSWORD
: The password to use with WSManPYPSRR_PORT
: The port to connect with over WSMan (default:5985
)PYPSRP_AUTH
: The authentication protocol to auth with (default:negotiate
)PYPSRP_SSH_SERVER
: The hostname or IP of the remote host to test SSH withPYPSRP_SSH_USERNAME
: The username to use with SSHPYPSRP_SSH_PASSWORD
: The password to use with SSHPYPSRP_SSH_KEY_PATH
: The path to a private key to use with SSHPYPSRP_SSH_IS_WINDOWS
: The remote SSH target is a Windows host