Welcome to PyNinja’s documentation!¶

Contents:

PyNinja

PyNinja - Main¶

pyninja.main.get_desc(get_api: bool, post_api: bool, monitoring_ui: bool) → str¶

Construct a detailed description for the API docs.

Parameters:

get_api – Boolean flag to indicate basic API state.
post_api – Boolean flag to indicate remote execution state.
monitoring_ui – Boolean flag to indicate monitoring page state.

Returns:

Returns the description as a string.

Return type:

str

async pyninja.main.redirect_exception_handler(request: Request, exception: RedirectException) → JSONResponse¶

Custom exception handler to handle redirect.

Parameters:

request – Takes the Request object as an argument.
exception – Takes the RedirectException object inherited from Exception as an argument.

Returns:

Returns the JSONResponse with content, status code and cookie.

Return type:

JSONResponse

pyninja.main.start(**kwargs) → None¶

Starter function for the API, which uses uvicorn server as trigger.

Keyword Arguments:

env_file – Env filepath to load the environment variables.
apikey – API Key for authentication.
ninja_host – Hostname for the API server.
ninja_port – Port number for the API server.
remote_execution – Boolean flag to enable remote execution.
api_secret – Secret access key for running commands on server remotely.
monitor_username – Username to authenticate the monitoring page.
monitor_password – Password to authenticate the monitoring page.
monitor_session – Session timeout for the monitoring page.
service_manager – Service manager filepath to handle the service status requests.
database – FilePath to store the auth database that handles the authentication errors.
rate_limit – List of dictionaries with max_requests and seconds to apply as rate limit.
log_config – Logging configuration as a dict or a FilePath. Supports .yaml/.yml, .json or .ini formats.

PyNinja - Executors¶

API Authenticator¶

pyninja.executors.auth.EPOCH()¶

async pyninja.executors.auth.forbidden(request: Request) → None¶

Validates if a request is part of the forbidden list.

Parameters:

request – Reference to the FastAPI request object.

Raises:

APIResponse –
- 403 – If host address is forbidden.

async pyninja.executors.auth.level_1(request: Request, apikey: HTTPAuthorizationCredentials) → None¶

Validates the auth request using HTTPBearer.

Parameters:

request – Takes the authorization header token as an argument.
apikey – Basic APIKey required for all the routes.

Raises:

APIResponse –
- 401 – If authorization is invalid.
- 403 – If host address is forbidden.

async pyninja.executors.auth.level_2(request: Request, apikey: HTTPAuthorizationCredentials, token: str) → None¶

Validates the auth request using HTTPBearer and additionally a secure token.

Parameters:

request – Takes the authorization header token as an argument.
apikey – Basic APIKey required for all the routes.
token – Additional token for critical requests.

Raises:

APIResponse –
- 401 – If authorization is invalid.
- 403 – If host address is forbidden.

async pyninja.executors.auth.incrementer(attempt: int) → int¶

Increments block time for a host address based on the number of failed login attempts.

Parameters:: attempt – Number of failed login attempts.
Returns:: Returns the appropriate block time in minutes.
Return type:: int

async pyninja.executors.auth.handle_auth_error(request: Request) → None¶

Handle authentication errors from the filebrowser API.

Parameters:: request – The incoming request object.

Database¶

pyninja.executors.database.get_record(host: str) → int | None¶

Gets blocked epoch time for a particular host.

Parameters:: host – Host address.
Returns:: Returns the epoch time until when the host address should be blocked.
Return type:: int

pyninja.executors.database.put_record(host: str, block_until: int) → None¶

Inserts blocked epoch time for a particular host.

Parameters:

host – Host address.
block_until – Epoch time until when the host address should be blocked.

pyninja.executors.database.remove_record(host: str) → None¶

Deletes all records related to the host address.

Parameters:: host – Host address.

API Routes¶

async pyninja.routes.fullaccess.run_command(request: Request, payload: RunCommand, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

API function to run a command on host machine.

Args:

request: Reference to the FastAPI request object.

payload: Payload received as request body.

apikey: API Key to authenticate the request.

token: API secret to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.fullaccess.list_files(request: Request, payload: ListFiles, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

Get all YAML files from fileio and all log files from logs directory.

Args:

request: Reference to the FastAPI request object.

payload: Payload received as request body.

apikey: API Key to authenticate the request.

token: API secret to authenticate the request.

Returns:

Dict[str, List[str]]: Dictionary of files that can be downloaded or uploaded.

async pyninja.routes.fullaccess.get_file(request: Request, payload: GetFile, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

Download a particular YAML file from fileio or log file from logs directory.

Args:

request: Reference to the FastAPI request object.

payload: Payload received as request body.

apikey: API Key to authenticate the request.

token: API secret to authenticate the request.

Returns:

FileResponse: Returns the FileResponse object of the file.

async pyninja.routes.fullaccess.put_file(request: Request, file: UploadFile, directory: Path, overwrite: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

Upload a file to th.

Args:

request: Reference to the FastAPI request object.

file: Upload object for the file param.

payload: Payload received as request body.

apikey: API Key to authenticate the request.

token: API secret to authenticate the request.

async pyninja.routes.ipaddr.get_ip_address(request: Request, public: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get local and public IP address of the device.

Args:

request: Reference to the FastAPI request object.

public: Boolean flag to get the public IP address.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and the public/private IP as response.

async pyninja.routes.metrics.get_cpu_utilization(request: Request, interval: int | float = 2, per_cpu: bool = True, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get the CPU utilization.

Args:

request: Reference to the FastAPI request object.

interval: Interval to get the CPU utilization.

per_cpu: If True, returns the CPU utilization for each CPU.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.

async pyninja.routes.metrics.get_memory_utilization(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get memory utilization.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.

async pyninja.routes.metrics.get_cpu_load_avg(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get the number of processes in the system run queue averaged over the last 1, 5, and 15 minutes respectively.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.

async pyninja.routes.metrics.get_disk_utilization(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get disk utilization.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.

async pyninja.routes.metrics.get_all_disks(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get all disks attached to the host device.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and attached disks as response.

async pyninja.routes.namespace.get_process_status(request: Request, process_name: str, cpu_interval: Union[int, float] = 1, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to monitor a process.

Args:

request: Reference to the FastAPI request object.

process_name: Name of the process to check status.

cpu_interval: Interval in seconds to get the CPU usage.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.namespace.get_service_usage(request: Request, service_names: List[str], apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to monitor a service.

Args:

request: Reference to the FastAPI request object.

service_names: Name of the service to check status.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.namespace.get_process_usage(request: Request, process_names: List[str], apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to monitor a process.

Args:

request: Reference to the FastAPI request object.

process_names: Name of the service to check status.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.namespace.get_service_status(request: Request, service_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to get the status of a service.

Args:

request: Reference to the FastAPI request object.

service_name: Name of the service to check status.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.namespace.get_processor_name(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to get process information.

Args:

request: Reference to the FastAPI request object.

process_name: Name of the process to get information.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.orchestration.get_docker_containers(request: Request, container_name: str = None, get_all: bool = False, get_running: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to get docker containers’ information.

Args:

request: Reference to the FastAPI request object.

container_name: Name of the container to check status.

get_all: Get all the containers’ information.

get_running: Get running containers’ information.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.orchestration.get_docker_images(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to get docker images’ information.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.orchestration.get_docker_volumes(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to get docker volumes’ information.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and detail as response.

async pyninja.routes.orchestration.get_docker_stats(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

Get docker-stats for all running containers.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code and attached disks as response.

async pyninja.executors.routers.monitor_redirect() → RedirectResponse¶

Redirect to monitor page.

Returns:: Redirects the user to /monitor page.
Return type:: RedirectResponse

async pyninja.executors.routers.docs_redirect() → RedirectResponse¶

Redirect to docs page.

Returns:: Redirects the user to /docs page.
Return type:: RedirectResponse

async pyninja.executors.routers.health()¶

Health check for PyNinja.

Returns:: Returns a health check response with status code 200.
Return type:: APIResponse

pyninja.executors.routers.get_api(dependencies: List[Depends]) → List[APIRoute]¶

Get all the routes to be added for the API server.

Parameters:: dependencies – List of dependencies to be added to the routes
Returns:: Returns the routes as a list of APIRoute objects.
Return type:: List[APIRoute]

pyninja.executors.routers.post_api(dependencies: List[Depends]) → List[APIRoute]¶

Get all the routes for FileIO operations and remote execution.

Parameters:: dependencies – List of dependencies to be added to the routes
Returns:: Returns the routes as a list of APIRoute objects.
Return type:: List[APIRoute]

pyninja.executors.routers.monitoring_ui(dependencies: List[Depends]) → List[fastapi.routing.APIRoute | fastapi.routing.APIWebSocketRoute]¶

Get all the routes for the monitor application.

Parameters:: dependencies – List of dependencies to be injected into the routes.
Returns:: Returns a list of API routes and WebSocket routes.
Return type:: List[APIRoute | APIWebSocketRoute]

Squire¶

pyninja.executors.squire.public_ip_address() → str¶

Gets public IP address of the host using different endpoints.

Returns:: Public IP address.
Return type:: str

pyninja.executors.squire.private_ip_address() → str | None¶

Uses a simple check on network id to see if it is connected to local host or not.

Returns:: Private IP address of host machine.
Return type:: str

pyninja.executors.squire.format_nos(input_: float) → int | float¶

Removes .0 float values.

Parameters:: input_ – Strings or integers with .0 at the end.
Returns:: Int if found, else returns the received float value.
Return type:: int | float

pyninja.executors.squire.format_timedelta(td: timedelta) → str¶

Converts timedelta to human-readable format by constructing a formatted string based on non-zero values.

Parameters:: td – Timedelta object.

See also

Always limits the output to a maximum of two identifiers.

Examples

3 days and 1 hour
1 hour and 11 minutes
5 minutes and 23 seconds

Returns:: Human-readable format of timedelta.
Return type:: str

pyninja.executors.squire.size_converter(byte_size: int | float) → str¶

Gets the current memory consumed and converts it to human friendly format.

Parameters:: byte_size – Receives byte size as argument.
Returns:: Converted human-readable size.
Return type:: str

pyninja.executors.squire.process_command(command: str, timeout: Union[int, float]) → Dict[str, List[str]]¶

Process the requested command.

Parameters:

command – Command as string.
timeout – Timeout for the command.

Returns:

Returns the result with stdout and stderr as key-value pairs.

Return type:

Dict[str, List[str]]

pyninja.executors.squire.envfile_loader(filename: str | os.PathLike) → EnvConfig¶

Loads environment variables based on filetypes.

Parameters:: filename – Filename from where env vars have to be loaded.
Returns:: Returns a reference to the EnvConfig object.
Return type:: EnvConfig

pyninja.executors.squire.load_env(**kwargs) → EnvConfig¶

Merge env vars from env_file with kwargs, giving priority to kwargs.

See also

This function allows env vars to be loaded partially from .env files and partially through kwargs.

Returns:: Returns a reference to the EnvConfig object.
Return type:: EnvConfig

pyninja.executors.squire.keygen() → str¶

Generate session token from secrets module, so that users are forced to log in when the server restarts.

Returns:: Returns a URL safe 64-bit token.
Return type:: str

pyninja.executors.squire.dynamic_numbers(string: str) → int | float | None¶

Convert strings to integer or float dynamically.

Parameters:: string – Number in string format.
Returns:: Integer or float value.
Return type:: int | float

pyninja.executors.squire.assert_pyudisk() → None¶: Ensure disk_report is enabled only for Linux machines and load udiskctl library.

pyninja.executors.squire.assert_tokens() → None¶: Ensure at least any of apikey or monitor username and monitor password is set.

pyninja.executors.squire.handle_warnings() → None¶: Raises security warnings.

pyninja.executors.squire.comma_separator(list_: list) → str¶

Separates commas using simple .join() function and includes and based on input length.

Parameters:: list_ – Takes a list of elements as an argument.
Returns:: Comma separated list of elements.
Return type:: str

pyninja.executors.squire.convert_seconds(seconds: int, n_elem: int = 2) → str¶

Calculate years, months, days, hours, minutes, and seconds from given input.

Parameters:

seconds – Number of seconds to convert.
n_elem – Number of elements required from the converted list.

Returns:

Returns a humanized string notion of the number of seconds.

Return type:

str

PyNinja - Features¶

CPU¶

pyninja.features.cpu._darwin() → str¶: Get processor information for macOS.

pyninja.features.cpu._linux() → str¶: Get processor information for Linux.

pyninja.features.cpu._windows() → str¶: Get processor information for Windows.

pyninja.features.cpu.get_name() → str | None¶

Get processor information for the host operating system.

Returns:: Returns the processor information as a string.
Return type:: str

Disks¶

pyninja.features.disks.__init__.get_all_disks() → List[Dict[str, str]]¶: OS-agnostic function to get all disks connected to the host system.

Disks - Linux¶

pyninja.features.disks.linux.drive_info() → List[Dict[str, str]]¶

Get disks attached to Linux devices.

Returns:: Returns disks information for Linux distros.
Return type:: List[Dict[str, str]]

Disks - macOS¶

pyninja.features.disks.macOS.parse_size(input_string: str) → int¶

Extracts size in bytes from a string.

Parameters:: input_string – Input string from diskutil output.
Returns:: Returns the size in bytes as an integer.
Return type:: int

pyninja.features.disks.macOS.update_mountpoints(disks: List[Dict[str, str]], device_ids: defaultdict) → defaultdict¶

Updates mount points for physical devices based on diskutil data.

Parameters:

disks – All disk info data as list.
device_ids – Device IDs as default dict.

Returns:

Returns a defaultdict object with updated mountpoints as list.

Return type:

defaultdict

pyninja.features.disks.macOS.parse_diskutil_output(stdout: str) → List[Dict[str, str]]¶

Parses diskutil info -all output into structured data.

Parameters:: stdout – Standard output from diskutil command.
Returns:: Returns a list of dictionaries with parsed drives’ data.
Return type:: List[Dict[str, str]]

pyninja.features.disks.macOS.drive_info() → List[Dict[str, str]]¶

Get disks attached to macOS devices.

Returns:: Returns disks information for macOS devices.
Return type:: List[Dict[str, str]]

Disks - Windows¶

pyninja.features.disks.windows.reformat_windows(data: Dict[str, str | int | float]) → Dict[str, str]¶

Reformats each drive’s information for Windows OS.

Parameters:: data – Data as a dictionary.
Returns:: Returns a dictionary of key-value pairs.
Return type:: Dict[str, str]

pyninja.features.disks.windows.get_drives() → List[Dict[str, str]]¶

Get physical drives connected to a Windows machine.

Returns:: Returns the formatted data for all the drives as a list of key-value pairs.
Return type:: List[Dict[str, str]]

pyninja.features.disks.windows.clean_ansi_escape_sequences(text: str) → str¶

Regular expression to remove ANSI escape sequences.

Parameters:: text – Text with ansi escape characters.
Returns:: Cleaned text.
Return type:: str

pyninja.features.disks.windows.get_physical_disks_and_partitions() → List[Tuple[str, str, str]]¶

Powershell Core command to get physical disks and their partitions with drive letters (mount points).

Returns:: List of tuples with disk_number, partition_number, mount_point.
Return type:: List[Tuple[str, str, str]]

pyninja.features.disks.windows.get_disk_usage() → Dict[str, List[str]]¶

Get all physical disks and their partitions with mount points.

Returns:: Returns a dictionary of DeviceID as key and mount paths as value.
Return type:: Dict[str, List[str]]

pyninja.features.disks.windows.drive_info() → List[Dict[str, str]]¶

Get disks attached to Windows devices.

Returns:: Returns disks information for Windows machines.
Return type:: List[Dict[str, str]]

Docker¶

pyninja.features.dockerized.get_container_status(name: str = None) → str | None¶

Get container status by name.

Parameters:: name – Name of the container or image used to run the container.
Returns:: Container status as a string.
Return type:: str

pyninja.features.dockerized.get_running_containers() → Generator[Dict[str, str]]¶

Get running containers.

Yields:: Dict[str, str] – Yields a dictionary of running containers with the corresponding metrics.

pyninja.features.dockerized.get_all_containers() → Optional[List[Dict[str, str]]]¶

Get all containers and their metrics.

Returns:: Returns a list of all the containers and their stats.
Return type:: List[Dict[str, str]]

pyninja.features.dockerized.get_all_images() → Optional[Dict[str, str]]¶

Get all docker images.

Returns:: Returns a dictionary with image stats.
Return type:: Dict[str, str]

pyninja.features.dockerized.get_all_volumes() → Optional[Dict[str, str]]¶

Get all docker volumes.

Returns:: Returns a dictionary with list of volume objects.
Return type:: Dict[str, str]

GPU¶

pyninja.features.gpu._darwin() → Optional[List[Dict[str, str]]]¶

Get GPU model and vendor information for Linux operating system.

Returns:: Returns a list of GPU model and vendor information.
Return type:: List[Dict[str, str]]

pyninja.features.gpu._linux() → Optional[List[Dict[str, str]]]¶

Get GPU model and vendor information for Linux operating system.

Returns:: Returns a list of GPU model and vendor information.
Return type:: List[Dict[str, str]]

pyninja.features.gpu._windows() → Optional[List[Dict[str, str]]]¶

Get GPU model and vendor information for Windows operating system.

Returns:: Returns a list of GPU model and vendor information.
Return type:: List[Dict[str, str]]

pyninja.features.gpu.get_names() → List[Dict[str, str]]¶: Get list of GPU model and vendor information based on the operating system.

Operations¶

pyninja.features.operations.default(name: str)¶: Default values for processes and services.

pyninja.features.operations.get_process_info(proc: Process, process_name: str = None) → Dict[str, Union[str, int]]¶

Get process information.

Parameters:

proc – Takes a psutil.Process object as an argument.
process_name – Takes a custom process name as an optional argument.

Returns:

Returns a dictionary with process usage statistics.

Return type:

Dict[str, str | int]

async pyninja.features.operations.process_monitor(processes: List[str]) → List[Dict[str, str]]¶

Function to monitor processes and return their usage statistics.

Parameters:: processes – List of process names to monitor.

See also

Process names can be case in-sensitive as they are not strictly matched.

macOS/Linux: top | grep {{ process_name }}

Windows: Task Manager

Returns:: Returns a list of dictionaries with process usage statistics.
Return type:: List[Dict[str, str]]

async pyninja.features.operations.service_monitor(services: List[str]) → List[Dict[str, str]]¶

Function to monitor services and return their usage statistics.

Parameters:: services – List of service names to monitor.

See also

Service names are case-sensitive as they are strictly matched. Use the following command to get the right name.

macOS: launchctl list | grep {{ service_name }}

Linux: systemctl show {{ service_name }} –property=MainPID

Windows: sc query {{ service_name }}

Returns:: Returns a list of dictionaries with service usage statistics.
Return type:: List[Dict[str, str]]

pyninja.features.operations.get_service_pid(service_name: str) → Optional[int]¶: Retrieve the PID of a service depending on the OS.

pyninja.features.operations.get_service_pid_linux(service_name: str) → Optional[int]¶

Get the PID of a service on Linux.

Parameters:: service_name – Name of the service.
Returns:: Returns the PID of the service.
Return type:: Optional[int]

pyninja.features.operations.get_service_pid_macos(service_name: str) → Optional[int]¶

Get the PID of a service on macOS.

Parameters:: service_name – Name of the service.
Returns:: Returns the PID of the service.
Return type:: Optional[int]

pyninja.features.operations.get_service_pid_windows(service_name: str) → Optional[int]¶

Get the PID of a service on Windows.

Parameters:: service_name – Name of the service.
Returns:: Returns the PID of the service.
Return type:: Optional[int]

Process¶

pyninja.features.process.get_process_status(process_name: str, cpu_interval: int) → List[Dict[str, int | float | str | bool]]¶

Get process information by name.

Parameters:

process_name – Name of the process.
cpu_interval – CPU interval to get the CPU performance.

Returns:

Returns a list of performance report for each process hosting the given process name.

Return type:

List[Dict[str, int | float | str | bool]]

pyninja.features.process.get_performance(process: Process, cpu_interval: int) → Dict[str, int | float | str | bool]¶

Checks process performance by monitoring CPU utilization, number of threads and open files.

Parameters:

process – Process object.
cpu_interval – CPU interval to get the CPU performance.

Returns:

Returns the process metrics as key-value pairs.

Return type:

Dict[str, int | float | str | bool]

Service¶

pyninja.features.service.running(service_name: str) → ServiceStatus¶

Constructs an ServiceStatus object with a status code.

Parameters:: service_name – Name of the service.
Returns:: Returns a reference to the ServiceStatus object.
Return type:: ServiceStatus

pyninja.features.service.stopped(service_name: str) → ServiceStatus¶

Constructs an ServiceStatus object with a status code 501.

Parameters:: service_name – Name of the service.
Returns:: Returns a reference to the ServiceStatus object.
Return type:: ServiceStatus

pyninja.features.service.unknown(service_name) → ServiceStatus¶

Constructs an ServiceStatus object with a status code 503.

Parameters:: service_name – Name of the service.
Returns:: Returns a reference to the ServiceStatus object.
Return type:: ServiceStatus

pyninja.features.service.unavailable(service_name: str) → ServiceStatus¶

Constructs an ServiceStatus object with a status code 404.

Parameters:: service_name – Name of the service.
Returns:: Returns a reference to the ServiceStatus object.
Return type:: ServiceStatus

pyninja.features.service.get_service_status(service_name: str) → ServiceStatus¶

Get service status by name.

Parameters:: service_name – Name of the service.
Returns:: Returns an instance of the ServiceStatus object.
Return type:: ServiceStatus

PyNinja - Modules¶

Cache¶

pyninja.modules.cache.timed_cache(max_age: int, maxsize: int = 128, typed: bool = False)¶

Least-recently-used cache decorator with time-based cache invalidation.

Parameters:

max_age – Time to live for cached results (in seconds).
maxsize – Maximum cache size (see functools.lru_cache).
typed – Cache on distinct input types (see functools.lru_cache).

See also

lru_cache takes all params of the function and creates a key.
If even one key is changed, it will map to new entry thus refreshed.
This is just a trick to force lru_cache lib to provide TTL on top of max size.
Uses time.monotonic since time.time relies on the system clock and may not be monotonic.
time.time() not always guaranteed to increase,

it may in fact decrease if the machine syncs its system clock over a network.

Enums¶

class pyninja.modules.enums.OperatingSystem(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶

Operating system names.

>>> OperatingSystem

linux: str = 'linux'¶

darwin: str = 'darwin'¶

windows: str = 'windows'¶

class pyninja.modules.enums.APIEndpoints(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶

API endpoints for all the routes.

>>> APIEndpoints

root: str = '/'¶

docs: str = '/docs'¶

redoc: str = '/redoc'¶

health: str = '/health'¶

get_file: str = '/get-file'¶

put_file: str = '/put-file'¶

get_processor: str = '/get-processor'¶

service_usage: str = '/service-usage'¶

process_usage: str = '/process-usage'¶

process_status: str = '/process-status'¶

service_status: str = '/service-status'¶

monitor: str = '/monitor'¶

ws_system: str = '/ws/system'¶

run_command: str = '/run-command'¶

get_disk: str = '/get-disk'¶

get_memory: str = '/get-memory'¶

get_all_disks: str = '/get-all-disks'¶

docker_image: str = '/docker-image'¶

docker_stats: str = '/docker-stats'¶

docker_volume: str = '/docker-volume'¶

docker_container: str = '/docker-container'¶

get_ip: str = '/get-ip'¶

get_cpu: str = '/get-cpu'¶

list_files: str = '/list-files'¶

get_cpu_load: str = '/get-cpu-load'¶

login: str = '/login'¶

logout: str = '/logout'¶

error: str = '/error'¶

class pyninja.modules.enums.Cookies(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶

Cookie names for the application.

>>> Cookies

drive: str = 'drive'¶

monitor: str = 'monitor'¶

session_token: str = 'session_token'¶

class pyninja.modules.enums.Templates(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)¶

HTML template filenames.

>>> Templates

main: str = 'main.html'¶

index: str = 'index.html'¶

logout: str = 'logout.html'¶

session: str = 'session.html'¶

disk_report: str = 'disk_report.html'¶

unauthorized: str = 'unauthorized.html'¶

Exceptions¶

exception pyninja.modules.exceptions.APIResponse(status_code: int, detail: Any = None, headers: Optional[Dict[str, str]] = None)¶

Custom HTTPException from FastAPI to wrap an API response.

>>> APIResponse

exception pyninja.modules.exceptions.UnSupportedOS¶

Custom exception class for unsupported OS.

>>> UnSupportedOS

exception pyninja.modules.exceptions.RedirectException(location: str, detail: Optional[str] = '')¶

Custom RedirectException raised within the API since HTTPException doesn’t support returning HTML content.

>>> RedirectException

See also

RedirectException allows the API to redirect on demand in cases where returning is not a solution.
There are alternatives to raise HTML content as an exception but none work with our use-case with JavaScript.
This way of exception handling comes handy for many unexpected scenarios.

References

https://fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers

exception pyninja.modules.exceptions.SessionError(detail: Optional[str] = '')¶

Custom exception class for session errors.

>>> SessionError

pyninja.modules.exceptions.raise_os_error(operating_system: str) → NoReturn¶

Raises a custom exception for unsupported OS.

Parameters:: operating_system – Current operating system.
Raises:: ValidationError – Overridden exception from pydantic.ValidationError for unsupported OS.

Models¶

class pyninja.modules.models.ServiceStatus(BaseModel)¶

Object to load service status with a status code and description.

>>> ServiceStatus

status_code: int¶

description: str¶

class pyninja.modules.models.DiskLib(BaseModel)¶

Default disks library dedicated to each supported operating system.

>>> DiskLib

linux: Path¶

darwin: Path¶

windows: Path¶

class pyninja.modules.models.ServiceLib(BaseModel)¶

Default service library dedicated to each supported operating system.

>>> ServiceLib

linux: Path¶

darwin: Path¶

windows: Path¶

class pyninja.modules.models.ProcessorLib(BaseModel)¶

Default processor library dedicated to each supported operating system.

>>> ProcessorLib

linux: Path¶

darwin: Path¶

windows: Path¶

class pyninja.modules.models.Session(BaseModel)¶

Object to store session information.

>>> Session

auth_counter: Dict[str, int]¶

forbid: Set[str]¶

info: Dict[str, str]¶

rps: Dict[str, int]¶

allowed_origins: Set[str]¶

class pyninja.modules.models.RateLimit(BaseModel)¶

Object to store the rate limit settings.

>>> RateLimit

max_requests: int¶

seconds: int¶

class pyninja.modules.models.EnvConfig(BaseModel)¶

Object to load environment variables.

>>> EnvConfig

apikey: str | None¶

ninja_host: str¶

ninja_port: int¶

workers: int¶

rate_limit: Union[RateLimit, List[RateLimit]]¶

log_config: Optional[Union[Dict[str, Any], Path]]¶

remote_execution: bool¶

api_secret: str | None¶

database: str¶

monitor_username: str | None¶

monitor_password: str | None¶

monitor_session: int¶

udisk_lib: Optional[Path]¶

disk_report: bool¶

max_connections: int¶

no_auth: bool¶

processes: List[str]¶

services: List[str]¶

gpu_lib: Path¶

disk_lib: Path¶

service_lib: Path¶

processor_lib: Path¶

classmethod parse_apikey(value: str | None) → str | None¶

Parse API key to validate complexity.

Parameters:: value – Takes the user input as an argument.
Returns:: Returns the parsed value.
Return type:: str

classmethod parse_api_secret(value: str | None) → str | None¶

Parse API secret to validate complexity.

Parameters:: value – Takes the user input as an argument.
Returns:: Returns the parsed value.
Return type:: str

classmethod from_env_file(env_file: Path) → EnvConfig¶

Create Settings instance from environment file.

Parameters:: env_file – Name of the env file.
Returns:: Loads the EnvConfig model.
Return type:: EnvConfig

class Config¶

Extra configuration for EnvConfig object.

extra = 'ignore'¶

hide_input_in_errors = True¶

pyninja.modules.models.complexity_checker(key: str, value: str, min_length: int) → None¶

Verifies the strength of a secret.

See also

A secret is considered strong if it at least has:

32 characters
1 digit
1 symbol
1 uppercase letter
1 lowercase letter

Raises:: AssertionError – When at least 1 of the above conditions fail to match.

class pyninja.modules.models.GPULib(*, linux: Path = '/usr/bin/lspci', darwin: Path = '/usr/sbin/system_profiler', windows: Path = 'C:\\Windows\\System32\\wbem\\wmic.exe')¶

Default GPU library dedicated to each supported operating system.

>>> GPULib

linux: Path¶

darwin: Path¶

windows: Path¶

_abc_impl = <_abc._abc_data object>¶

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'darwin': FieldInfo(annotation=Path, required=False, default='/usr/sbin/system_profiler', metadata=[PathType(path_type='file')]), 'linux': FieldInfo(annotation=Path, required=False, default='/usr/bin/lspci', metadata=[PathType(path_type='file')]), 'windows': FieldInfo(annotation=Path, required=False, default='C:\\Windows\\System32\\wbem\\wmic.exe', metadata=[PathType(path_type='file')])}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class pyninja.modules.models.WSSession(*, invalid: Dict[str, int] = None, client_auth: Dict[str, Dict[str, int]] = None)¶

Object to store websocket session information.

>>> WSSession

invalid: Dict[str, int]¶

client_auth: Dict[str, Dict[str, int]]¶

_abc_impl = <_abc._abc_data object>¶

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}¶: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'client_auth': FieldInfo(annotation=Dict[str, Dict[str, int]], required=False, default_factory=dict), 'invalid': FieldInfo(annotation=Dict[str, int], required=False, default_factory=dict)}¶

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

pyninja.modules.models.get_library(library: Union[Type[ServiceLib], Type[ProcessorLib], Type[DiskLib], Type[GPULib]]) → Path¶

Get service/processor/disk library filepath for the host operating system.

Parameters:: library – Library class inherited from BaseModel.
Returns:: Returns the FilePath referencing the appropriate library.
Return type:: FilePath

class pyninja.modules.models.Database(datastore: Union[Path, str], timeout: int = 10)¶

Creates a connection to the Database using sqlite3.

>>> Database

create_table(table_name: str, columns: Union[List[str], Tuple[str]]) → None¶

Creates the table with the required columns.

Parameters:

table_name – Name of the table that has to be created.
columns – List of columns that has to be created.

Payloads¶

class pyninja.modules.payloads.RunCommand(BaseModel)¶

Payload for run-command endpoint.

>>> RunCommand

command: str¶

timeout: Union[int, float]¶

class pyninja.modules.payloads.ListFiles(BaseModel)¶

Payload for list-files endpoint.

>>> ListFiles

directory: Path¶

show_hidden_files: bool¶

include_directories: bool¶

deep_scan: bool¶

class pyninja.modules.payloads.GetFile(BaseModel)¶

Payload for get-file endpoint.

>>> GetFile

filepath: Path¶

RateLimit¶

class pyninja.modules.rate_limit.RateLimiter(rps: RateLimit)¶

Object that implements the RateLimiter functionality.

>>> RateLimiter

init(request: Request) → None¶

Checks if the number of calls exceeds the rate limit for the given identifier.

Parameters:: request – The incoming request object.
Raises:: 429 – Too many requests.

Secure¶

async pyninja.modules.secure.calculate_hash(value: Any) → str¶: Generate hash value for the given payload.

async pyninja.modules.secure.base64_encode(value: Any) → str¶: Base64 encode the given payload.

async pyninja.modules.secure.base64_decode(value: Any) → str¶: Base64 decode the given payload.

async pyninja.modules.secure.hex_decode(value: Any) → str¶: Convert hex value to a string.

async pyninja.modules.secure.hex_encode(value: str)¶: Convert string value to hex.

PyNinja - Monitor¶

Authenticator¶

async pyninja.monitor.authenticator.failed_auth_counter(host) → None¶

Keeps track of failed login attempts from each host, and redirects if failed for 3 or more times.

Parameters:: host – Host header from the request.

async pyninja.monitor.authenticator.raise_error(host: str) → NoReturn¶

Raises a 401 Unauthorized error in case of bad credentials.

Parameters:: host – Host header from the request.

async pyninja.monitor.authenticator.extract_credentials(authorization: HTTPAuthorizationCredentials, host: str) → List[str]¶

Extract the credentials from Authorization headers and decode it before returning as a list of strings.

Parameters:

authorization – Authorization header from the request.
host – Host header from the request.

async pyninja.monitor.authenticator.verify_login(authorization: HTTPAuthorizationCredentials, host: str) → Dict[str, Union[str, int]]¶

Verifies authentication and generates session token for each user.

Returns:: Returns a dictionary with the payload required to create the session token.
Return type:: Dict[str, str]

async pyninja.monitor.authenticator.generate_cookie(auth_payload: dict) → Dict[str, str | bool | int]¶

Generate a cookie for monitoring page.

Parameters:: auth_payload – Authentication payload containing username and timestamp.
Returns:: Returns a dictionary with cookie details
Return type:: Dict[str, str | bool | int]

async pyninja.monitor.authenticator.session_error(request: Request, error: SessionError) → HTMLResponse¶

Renders the session error page.

Parameters:

request – Reference to the FastAPI request object.
error – Session error message.

Returns:

Returns an HTML response templated using Jinja2.

Return type:

HTMLResponse

async pyninja.monitor.authenticator.validate_session(host: str, cookie_string: str, log: bool = True) → None¶

Validate the session token.

Parameters:

host – Hostname from the request.
cookie_string – Session token from the cookie.
log – Boolean flag to enable logging.

Raises:

SessionError –
Raises a SessionError with summary. –

Configuration¶

async pyninja.monitor.config.clear_session(request: Request, response: HTMLResponse) → HTMLResponse¶

Clear the session token from the response.

Parameters:

request – FastAPI request object.
response – FastAPI response object.

Returns:

Returns the response object with the session token cleared.

Return type:

HTMLResponse

async pyninja.monitor.config.get_expiry(lease_start: int, lease_duration: int) → str¶

Get expiry datetime as string using max age.

Parameters:

lease_start – Time when the authentication was made.
lease_duration – Number of seconds until expiry.

Returns:

Returns the date and time of expiry in GMT.

Return type:

str

Drive¶

async pyninja.monitor.drive.report(request: Request) → HTMLResponse¶

Generates a disk report using pyudisk.

Returns:: Returns an HTML response with the disk report.
Return type:: HTMLResponse

async pyninja.monitor.drive.invalidate(ctx: str, status_code: int = 500) → HTMLResponse¶

Invalidates the cache after wrapping the context into HTMLResponse object.

Parameters:

ctx – Content to enter into the HTMLResponse object.
status_code – Status code for the response.

Returns:

Returns an HTML response with the given context and status code.

Return type:

HTMLResponse

Resources¶

pyninja.monitor.resources.landing_page() → Dict[str, Any]¶

Returns the landing page context for monitor endpoint.

Returns:: Returns a key-value pair to be inserted into the Jinja template.
Return type:: Dict[str, Any]

pyninja.monitor.resources.get_disk_info() → Generator[Dict[str, Union[str, int]]]¶

Get partition and usage information for each physical drive.

Yields:: Dict[str, str | int] – Yields a dictionary of key-value pairs with ID, name, and usage.

pyninja.monitor.resources.container_cpu_limit(container_name: str) → int | float | None¶

Get CPU cores configured for a particular container using NanoCpus.

Parameters:: container_name – Name of the docker container.
Returns:: Returns the number of CPU cores.
Return type:: int

pyninja.monitor.resources.map_docker_stats(json_data: Dict[str, str]) → Dict[str, str]¶

Map the JSON data to a dictionary.

Parameters:: json_data – JSON data from the docker stats command.
Returns:: Returns a dictionary with container stats.
Return type:: Dict[str, str]

pyninja.monitor.resources.get_cpu_percent(cpu_interval: int) → List[float]¶

Get CPU usage percentage.

Parameters:: cpu_interval – Interval to get the CPU performance.
Returns:: Returns a list of CPU percentages.
Return type:: List[float]

pyninja.monitor.resources.containers() → bool¶: Check if any Docker containers are running.

async pyninja.monitor.resources.get_docker_stats() → List[Dict[str, str]]¶

Run the docker stats command asynchronously and parse the output.

Returns:: Returns a list of key-value pairs with the container stat and value.
Return type:: List[Dict[str, str]]

async pyninja.monitor.resources.get_system_metrics() → Dict[str, dict]¶

Async handler for virtual memory, swap memory disk usage and CPU load averages.

Returns:: Returns a nested dictionary.
Return type:: Dict[str, dict]

pyninja.monitor.resources.pyudisk_metrics() → Dict[str, Union[str, List[dict]]]¶

Retrieves metrics from PyUdisk library.

See also

This is a timed-cache function. Meaning: The output from this function will be cached for 60s.
This is to avoid gathering the metrics every 2s, to improve latency and avoid extra overhead.

Returns:: List of required metrics as a dictionary of key-value pairs.
Return type:: List[Dict[str, int | str | float]]

async pyninja.monitor.resources.system_resources() → Dict[str, Union[dict, List[Dict[str, Union[str, int]]]]]¶

Gather system resources including Docker stats asynchronously.

Returns:: Returns a nested dictionary.
Return type:: Dict[str, dict]

Routes¶

async pyninja.monitor.routes.error_endpoint(request: Request) → HTMLResponse¶

Error endpoint for the monitoring page.

Parameters:: request – Reference to the FastAPI request object.
Returns:: Returns an HTML response templated using Jinja2.
Return type:: HTMLResponse

async pyninja.monitor.routes.logout_endpoint(request: Request) → HTMLResponse¶

Logs out the user and clears the session token.

Parameters:: request – Reference to the FastAPI request object.
Returns:: Redirects to login page.
Return type:: HTMLResponse

async pyninja.monitor.routes.login_endpoint(request: Request, authorization: HTTPAuthorizationCredentials = Depends(HTTPBearer)) → JSONResponse¶

Login endpoint for the monitoring page.

Returns:: Returns a JSONResponse object with a session_token and redirect_url set.
Return type:: JSONResponse

async pyninja.monitor.routes.monitor_endpoint(request: Request, session_token: str = Cookie(None), render: str = Cookie(None))¶

Renders the UI for monitoring page.

Parameters:

request – Reference to the FastAPI request object.
session_token – Session token set after verifying username and password.
render – Render option set by the UI.

Returns:

Returns an HTML response templated using Jinja2.

Return type:

HTMLResponse

async pyninja.monitor.routes.websocket_endpoint(websocket: WebSocket, session_token: str = Cookie(None))¶

Websocket endpoint to fetch live system resource usage.

Parameters:

websocket – Reference to the websocket object.
session_token – Session token set after verifying username and password.

Welcome to PyNinja’s documentation!¶

PyNinja - Main¶

PyNinja - Executors¶

API Authenticator¶

Database¶

API Routes¶

Squire¶

PyNinja - Features¶

CPU¶

Disks¶

Disks - Linux¶

Disks - macOS¶

Disks - Windows¶

Docker¶

GPU¶

Operations¶

Process¶

Service¶

PyNinja - Modules¶

Cache¶

Enums¶

Exceptions¶

Models¶

Payloads¶

RateLimit¶

Secure¶

PyNinja - Monitor¶

Authenticator¶

Configuration¶

Drive¶

Resources¶

Routes¶

Indices and tables¶

Table of Contents

Next topic

This Page