Welcome to PyNinja’s documentation!¶

Contents:

PyNinja

PyNinja - Main¶

pyninja.main.get_desc(remote_flag: bool, monitor_flag: bool) → str¶

Construct a detailed description for the API docs.

Parameters:

remote_flag – Boolean flag to indicate remote execution state.
monitor_flag – 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.docs() → 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_all_routes(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]

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 understandable 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 - Features¶

CPU¶

pyninja.features.cpu._darwin(lib_path: Path) → str¶

Get processor information for macOS.

Parameters:: lib_path – Path to the library executable.

pyninja.features.cpu._linux(lib_path: Path) → str¶

Get processor information for Linux.

Parameters:: lib_path – Path to the library file.

pyninja.features.cpu._windows(lib_path: Path) → str¶

Get processor information for Windows.

Parameters:: lib_path – Path to the library file.

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.get_partitions_for_disk(device_id: str) → List[str]¶

Use psutil to find partitions for a given device.

Parameters:: device_id
Returns:: Returns the list of partitions in a disk.
Return type:: List[str]

pyninja.features.disks.parse_size(size_str) → str¶

Convert size string with units to a standard size in bytes.

Parameters:: size_str – Size with unit as string.
Returns:: Returns the parsed size as string with units attached.
Return type:: str

pyninja.features.disks._linux(lib_path: Path) → List[Dict[str, str]]¶

Get disks attached to Linux devices.

Parameters:: lib_path – Returns the library path for disk information.
Returns:: Returns disks information for Linux distros.
Return type:: List[Dict[str, str]]

pyninja.features.disks.is_physical_disk(lib_path: Path, device_id: str) → bool¶

Check if the disk is a physical disk using diskutil info.

Parameters:

lib_path – Library path to get disk info.
device_id – Device mount ID.

Returns:

Boolean to indicate virtual or physical

Return type:

bool

pyninja.features.disks._darwin(lib_path: Path) → List[Dict[str, str]]¶

Get disks attached to macOS devices.

Parameters:: lib_path – Returns the library path for disk information.
Returns:: Returns disks information for macOS devices.
Return type:: List[Dict[str, str]]

pyninja.features.disks._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(lib_path: Path) → List[Dict[str, str]]¶

Get disks attached to Windows devices.

Parameters:: lib_path – Returns the library path for disk information.
Returns:: Returns disks information for Windows machines.
Return type:: List[Dict[str, str]]

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

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(lib_path) → Optional[List[Dict[str, str]]]¶

Get GPU model and vendor information for Linux operating system.

Parameters:: lib_path – Library path to get GPU information.
Returns:: Returns a list of GPU model and vendor information.
Return type:: List[Dict[str, str]]

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

Get GPU model and vendor information for Linux operating system.

Parameters:: lib_path – Library path to get GPU information.
Returns:: Returns a list of GPU model and vendor information.
Return type:: List[Dict[str, str]]

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

Get GPU model and vendor information for Windows operating system.

Parameters:: lib_path – Library path to get GPU information.
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, 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¶

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¶

ninja_host: str¶

ninja_port: int¶

remote_execution: bool¶

api_secret: str | None¶

monitor_username: str | None¶

monitor_password: str | None¶

monitor_session: int¶

max_connections: int¶

no_auth: bool¶

processes: List[str]¶

services: List[str]¶

gpu_lib: Path¶

disk_lib: Path¶

service_lib: Path¶

processor_lib: Path¶

database: str¶

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

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

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(secret: str) → 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] = {}, client_auth: Dict[str, Dict[str, int]] = {})¶

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={}), 'invalid': FieldInfo(annotation=Dict[str, int], required=False, default={})}¶

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, 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(response: HTMLResponse) → HTMLResponse¶

Clear the session token from the response.

Parameters:: response – Takes the Response object as an argument.
Returns:: Returns the response object with the session token cleared.
Return type:: Response

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

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.container_cpu_limit(container_name: str) → int | 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]

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

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))¶

Renders the UI for monitoring page.

Parameters:

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

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¶

Docker¶

GPU¶

Operations¶

Process¶

Service¶

PyNinja - Modules¶

Exceptions¶

Models¶

Payloads¶

RateLimit¶

Secure¶

PyNinja - Monitor¶

Authenticator¶

Configuration¶

Resources¶

Routes¶

Indices and tables¶

Table of Contents

Next topic

This Page