Welcome to PyNinja’s documentation!¶

Contents:

PyNinja

PyNinja - Main¶

async pyninja.main.docs() → HTMLResponse¶

Custom docs endpoint for the Swagger UI.

See also

The Swagger UI is customized to scroll to the operation when a hyperlink from the description block is selected.

Returns:: Returns an HTMLResponse object with the customized UI.
Return type:: HTMLResponse

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

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

Keyword Arguments:

Environment_variables_configuration –
- env_file: Env filepath to load the environment variables.
Basic_API_functionalities –
- apikey: API Key for authentication.
- swagger_ui_parameters: Parameters for the Swagger UI.
- ninja_host: Hostname for the API server.
- ninja_port: Port number for the API server.
Functional_improvements –
- 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.
Remote_execution_and_FileIO –
- remote_execution: Boolean flag to enable remote execution.
- api_secret: Secret access key for running commands on server remotely.
- database: FilePath to store the auth database that handles the authentication errors.
Monitoring_UI –
- monitor_username: Username to authenticate the monitoring page.
- monitor_password: Password to authenticate the monitoring page.
- monitor_session: Session timeout for the monitoring page.
- disk_report: Boolean flag to enable disk report generation.
- max_connections: Maximum number of connections to handle.
- no_auth: Boolean flag to disable authentication for monitoring page.
- processes: List of process names to include in the monitoring page.
- services: List of service names to include in the monitoring page.
- service_lib: Library path to retrieve service info.
- smart_lib: Library path for S.M.A.R.T metrics using PyUdisk.
- gpu_lib: Library path to retrieve GPU names using PyArchitecture.
- disk_lib: Library path to retrieve disk info using PyArchitecture.
- processor_lib: Library path to retrieve processor name using PyArchitecture.

pyninja.startup.docs_handler(api: FastAPI, func: Callable) → None¶

Removes the default Swagger UI endpoint and adds a custom docs endpoint.

Parameters:

api – FastAPI object to modify the routes.
func – Callable function to be used as the endpoint for the custom docs.

References

https://swagger.io/docs/open-source-tools/swagger-ui/usage/configuration/

pyninja.startup.generate_hyperlink(route: fastapi.routing.APIRoute | fastapi.routing.APIWebSocketRoute) → str¶

Generates hyperlink for a particular API route to be included in the description.

Parameters:: route – APIRoute or APIWebSocketRoute object.
Returns:: Returns the hyperlink as a string.
Return type:: str

pyninja.startup.get_desc(get_routes: RoutingHandler, post_routes: RoutingHandler, monitor_routes: RoutingHandler) → str¶

Construct a detailed description for the API docs.

Parameters:

get_routes – RoutingHandler object for GET endpoints.
post_routes – RoutingHandler object for POST endpoints.
monitor_routes – RoutingHandler object for MONITOR endpoints.

Returns:

Returns the description as a string.

Return type:

str

async pyninja.startup.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 - 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, path: str = '/', 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_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to monitor a service.

Args:

request: Reference to the FastAPI request object.

process_name: Comma separated list of service names.

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_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to monitor a process.

Args:

request: Reference to the FastAPI request object.

process_name: Comma separated list of process names.

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_all_services(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶

API function to get a list of all the services.

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.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.stop_service(request: Request, service_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

API function to stop 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.

token: API secret to authenticate the request.

Raises:

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

async pyninja.routes.namespace.start_service(request: Request, service_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

API function to start 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.

token: API secret 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.stop_docker_container(request: Request, container_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

API function to stop a docker container.

Args:

request: Reference to the FastAPI request object.

container_name: Name of the container to stop.

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.orchestration.start_docker_container(request: Request, container_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶

API function to start a docker container.

Args:

request: Reference to the FastAPI request object.

container_name: Name of the container to start.

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.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.load_architecture(env: EnvConfig) → Architecture¶

Load architecture details from environment variables.

Parameters:: env – Environment variables.
Returns:: Returns a reference to the Architecture object.
Return type:: Architecture

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.executors.squire.humanize_usage_metrics(**kwargs) → Dict[str, str]¶: Convert the usage metrics into human-readable format.

pyninja.executors.squire.total_mountpoints_usage(mountpoints: List[str], as_bytes: bool = False) → Dict[str, int | str]¶

Sums up the bytes used on all the mountpoint locations.

Parameters:

mountpoints – List of mountpoints.
as_bytes – Boolean flag to return the dict as bytes.

Returns:

Returns the usage dictionary as key-value pairs.

Return type:

Dict[str, int | str]

PyNinja - Features¶

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]

pyninja.features.dockerized.stop_container(container_name: str)¶

Stop a container by name.

Parameters:: container_name – Name of the container.
Returns:: Returns the status of the container.
Return type:: str

pyninja.features.dockerized.start_container(container_name: str)¶

Start a container by name.

Parameters:: container_name – Name of the container.
Returns:: Returns the status of the container.
Return type:: str

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, int | str]¶

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_process_object(pid: str, service: str) → psutil.Process | None¶

Creates a process object using the service PID.

Parameters:

pid – Process ID as a string.
service – Name of the service.

Returns:

Returns a reference to the psutil.Process object.

Return type:

psutil.Process

pyninja.features.service.get_all_services() → Generator[Dict[str, str]]¶

OS-agnostic function to list all the services available and their status.

Yields:: Dict[str, str] – Yields all the services as key-value pairs.

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.features.service.stop_service(service_name: str)¶

Stop a service by name.

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

pyninja.features.service.start_service(service_name: str)¶

Start a service 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'¶

get_all_services: str = '/get-all-services'¶

get_service_status: str = '/get-service-status'¶

get_service_usage: str = '/get-service-usage'¶

stop_service: str = '/stop-service'¶

start_service: str = '/start-service'¶

get_process_status: str = '/get-process-status'¶

get_process_usage: str = '/get-process-usage'¶

monitor: str = '/monitor'¶

ws_system: str = '/ws/system'¶

run_command: str = '/run-command'¶

get_memory: str = '/get-memory'¶

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

get_disk_utilization: str = '/get-disk-utilization'¶

get_docker_images: str = '/get-docker-images'¶

get_docker_stats: str = '/get-docker-stats'¶

get_docker_volumes: str = '/get-docker-volumes'¶

get_docker_containers: str = '/get-docker-containers'¶

stop_docker_container: str = '/stop-docker-container'¶

start_docker_container: str = '/start-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'¶

unauthorized: str = 'unauthorized.html'¶

disk_report_linux: str = 'disk_report_linux.html'¶

disk_report_darwin: str = 'disk_report_darwin.html'¶

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

Types of API routes available.

>>> APIRouteType

get: str = 'get'¶

post: str = 'post'¶

monitor: str = 'monitor'¶

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.RoutingHandler(BaseModel)¶

Routing handler to update the API routes to the server.

>>> RoutingHandler

type: APIRouteType¶

routes: List[fastapi.routing.APIRoute | fastapi.routing.APIWebSocketRoute]¶

enabled: bool¶

class Config¶

Configuration for routing handler.

arbitrary_types_allowed = True¶

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

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

>>> ServiceStatus

status_code: int¶

description: str¶

service_name: str¶

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

Object to store the architecture of the system.

>>> Architecture

cpu: str¶

gpu: List[Dict[str, str]]¶

disks: List[Dict[str, Any]]¶

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.WSSession(BaseModel)¶

Object to store websocket session information.

>>> WSSession

invalid: Dict[str, int]¶

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

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

Object to store the rate limit settings.

>>> RateLimit

max_requests: int¶

seconds: int¶

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

Object to load the file I/O settings.

>>> FileIO

swagger_ui: str¶

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

Object to load environment variables.

>>> EnvConfig

apikey: str | None¶

swagger_ui_parameters: Dict[str, Any]¶

ninja_host: str¶

ninja_port: 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¶

disk_report: bool¶

max_connections: int¶

no_auth: bool¶

processes: List[str]¶

services: List[str]¶

smart_lib: Optional[Path]¶

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.

pyninja.modules.models.default_service_lib() → Path¶

Get default service library filepath for the host operating system.

Returns:: Returns the FilePath referencing the appropriate library.
Return type:: FilePath

pyninja.modules.models.retrieve_library_path(func: Callable) → Path¶

Retrieves the library path from the mapping created for each operating system.

Parameters:: func – Function to call to get the mapping.
Returns:: Returns the library path as a FilePath object.
Return type:: FilePath

pyninja.modules.models.load_swagger_ui() → str¶: Get the custom JavaScript for Swagger UI.

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, int | str]¶

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, int | str]]¶

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.get_os_agnostic_metrics() → Generator[Dict[str, Any]]¶

Retrieves OS-agnostic PyUdisk metrics.

Yields:: Dict[str, Any] – Yields a dictionary of retrieved values.

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

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, int | str]]]]¶

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¶

Docker¶

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