Welcome to PyNinja’s documentation!

Contents:

PyNinja - Main

pyninja.main.lifespan(app: FastAPI)

Lifespan context manager to handle startup and shutdown events.

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/

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.

pyninja.executors.auth.invalid_mfa() → NoReturn

Raises an APIResponse for invalid or expired MFA code.

async pyninja.executors.auth.verify_mfa(mfa_code: str) → None

Verifies the multifactor authentication code.

Parameters:

mfa_code – Multifactor authentication code to verify.

Raises:
async pyninja.executors.auth.level_2(request: Request, apikey: HTTPAuthorizationCredentials, api_secret: str, mfa_code: 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.

  • api_secret – API secret for secured endpoints.

  • mfa_code – Multifactor authentication code for additional security.

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.

Multifactor Authentication

pyninja.executors.multifactor.instantiate_mailer() → SendEmail

Cached function to instantiate the gmail-connector object.

Returns:

Returns an instance of the SendEmail class from the gmail-connector module.

Return type:

gc.SendEmail

pyninja.executors.multifactor.reset_mfa_code() → None

Resets the stored MFA token after the timeout period.

pyninja.executors.multifactor.clear_timer_list() → None

Clears the list of active timers.

pyninja.executors.multifactor.send_new_mfa() → bool

Function to check if a new MFA token should be sent.

async pyninja.executors.multifactor.get_mfa(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

Get multifactor authentication code.

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 the public/private IP as response.

API Routes

pyninja.routes.fullaccess.create_directory(directory: Path) → Union[None, NoReturn]

Create a directory if it does not exist.

Parameters:

directory – Directory path to create.

Raises:

  • APIResponse –

  • Raises an INTERNAL_SERVER_ERROR if the directory cannot be created. –

async pyninja.routes.fullaccess.run_command(request: Request, payload: RunCommand, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: 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.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

Raises:

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

async pyninja.routes.fullaccess.delete_content(request: Request, payload: DeleteContent, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

List files in a directory or scan the directory tree.

Args:

  • request: Reference to the FastAPI request object.

  • payload: Payload received as request body.

  • apikey: API Key to authenticate the request.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

Returns:

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

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

List files in a directory or scan the directory tree.

Args:

  • request: Reference to the FastAPI request object.

  • payload: Payload received as request body.

  • apikey: API Key to authenticate the request.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

Download a particular file.

Args:

  • request: Reference to the FastAPI request object.

  • payload: Payload received as request body.

  • apikey: API Key to authenticate the request.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

Upload a file to th.

Args:

  • request: Reference to the FastAPI request object.

  • file: Upload object for the file param.

  • directory: Target directory for upload.

  • overwrite: Boolean flag to remove existing file.

  • payload: Payload received as request body.

  • apikey: API Key to authenticate the request.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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), api_secret: Optional[str] = Header(None), mfa_code: 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.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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), api_secret: Optional[str] = Header(None), mfa_code: 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.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

Raises:

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

async pyninja.routes.namespace.restart_service(request: Request, service_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

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

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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.namespace.get_all_apps(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))

API function to get a list of all the applications.

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.

pyninja.routes.namespace.unavailable(app_name: str) → NoReturn

Return an APIResponse indicating the application is unavailable, along with a list of available applications.

async pyninja.routes.namespace.start_application(request: Request, app_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

API function to start an application.

Args:

request: Reference to the FastAPI request object. app_name: Name of the application to start. apikey: API Key to authenticate the request. api_secret: API secret to authenticate the request.

Raises:

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

async pyninja.routes.namespace.stop_application(request: Request, app_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

API function to stop an application.

Args:

request: Reference to the FastAPI request object. app_name: Name of the application to stop. apikey: API Key to authenticate the request. api_secret: API secret to authenticate the request.

Raises:

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

async pyninja.routes.namespace.restart_application(request: Request, app_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

API function to restart an application.

Args:

request: Reference to the FastAPI request object. app_name: Name of the application to restart. apikey: API Key to authenticate the request. api_secret: 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_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), api_secret: Optional[str] = Header(None), mfa_code: 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.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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), api_secret: Optional[str] = Header(None), mfa_code: 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.

  • api_secret: API secret to authenticate the request.

  • mfa_code: Multifactor authentication code.

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]

API Routes - Large Scale

Download

Module to handle large file downloads in chunks via FastAPI with optional server-side compression.

pyninja.routes.download.iter_file_chunks(filepath: Path, chunk_size: int) → Generator[bytes]

Generator to read a file in chunks.

Parameters:

  • filepath – FilePath to read.

  • chunk_size – Chunk size to read the file in bytes. Default is 8 KB.

Yields:

bytes – Yields chunks of the file as bytes.

async pyninja.routes.download.get_large_file(request: Request, filepath: Optional[Path] = None, directory: Optional[Path] = None, chunk_size: int = 8192, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

API handler to download a large file or directory as a compressed archive.

Parameters:

  • - request – Reference to the FastAPI request object.

  • - filepath – FilePath to the file to download.

  • - directory – DirectoryPath to compress and download.

  • - apikey – API Key to authenticate the request.

  • - api_secret – API secret to authenticate the request.

  • - mfa_code – Multifactor authentication code.

Returns:

Returns the FileResponse object of the file.

Return type:

FileResponse

Upload

Module to handle large file uploads in chunks via FastAPI with optional server-side unzip and checksum validation.

async pyninja.routes.upload.entry_fn(filename: str, filepath: str, tmp_filepath: str, directory: str, overwrite: bool = False, unzip: bool = False) → None

Entry function to handle the initial setup for large file upload.

async pyninja.routes.upload.exit_fn(filename: str, filepath: str, tmp_filepath: str, directory: str, checksum: Optional[str], unzip: bool, delete_after_unzip: bool, iteration: int) → None

Exit function to finalize the large file upload.

async pyninja.routes.upload.put_large_file(request: Request, filename: str, directory: str, part_number: int = 0, is_last: bool = False, checksum: Optional[str] = None, overwrite: bool = False, unzip: bool = False, delete_after_unzip: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), api_secret: Optional[str] = Header(None), mfa_code: Optional[str] = Header(None))

API handler to upload a large file in chunks.

Parameters:

  • - request – Reference to the FastAPI request object.

  • - filename – Incoming file’s basename.

  • - directory – Target directory for upload.

  • - part_number – Incoming file part number.

  • - is_last – Boolean flag to indicate that the incoming chunk is final.

  • - checksum – Incoming file checksum.

  • - overwrite – Boolean flag to remove existing file.

  • - unzip – Boolean flag to unzip the file after upload.

  • - delete_after_unzip – Boolean flag to delete the file after unzipping.

  • - apikey – API Key to authenticate the request.

  • - api_secret – API secret to authenticate the request.

  • - mfa_code – Multifactor authentication code.

PyNinja - Features

Application

Special module for managing GUI applications on macOS.

>>> macOS - Applications
pyninja.features.application.get_all_apps() → Generator[Dict[str, str]]

Get all GUI applications installed in the /Applications directory.

Yields:

Dict[str, str] – Yields a dictionary with the display name as the key and the application path as the value.

pyninja.features.application.is_app_running(app_name: str) → bool

Check if a GUI application is currently running.

Parameters:

app_name – Name of the application to check.

Returns:

Returns True if the application is running, otherwise False.

Return type:

bool

pyninja.features.application.unavailable(app_name: str) → AppStatus

Return an AppStatus indicating the application is unavailable.

Parameters:

app_name – Name of the application.

Returns:

Returns an instance of the AppStatus object with status code 404.

Return type:

AppStatus

pyninja.features.application.failed(app_name: str, app_path: Path, error: str) → AppStatus

Return an AppStatus indicating the application failed to restart.

Parameters:

  • app_name – Name of the application.

  • app_path – Application path.

  • error – Error message describing the failure.

Returns:

Returns an instance of the AppStatus object with status code 500.

Return type:

AppStatus

pyninja.features.application.success(app_name: str, app_path: str) → AppStatus

Return an AppStatus indicating the application restarted successfully.

Parameters:

  • app_name – Name of the application.

  • app_path – Application path.

Returns:

Returns an instance of the AppStatus object with status code 200.

Return type:

AppStatus

pyninja.features.application.stop_app(app_path: str) → None

Stop a GUI service by name.

Parameters:

app_path – Path to the application.

pyninja.features.application.start_app(app_path: str) → None

Start a GUI service by path.

Parameters:

app_path – Application path to start.

pyninja.features.application.get_app_by_name(app_name: str) → Dict[str, str]

Get the application path by its name.

Parameters:

app_name – Name of the application.

Returns:

Returns a dictionary with the application name as the key and its path as the value.

Return type:

Dict[str, str]

pyninja.features.application.restart(app_name: str) → AppStatus

Restart a GUI service by name.

Parameters:

app_name – Name of the application.

Returns:

Returns the path to the application if restarted successfully, otherwise None.

Return type:

str | None

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) → ServiceStatus

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) → ServiceStatus

Start 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.restarted(service_name: str) → ServiceStatus

Constructs an ServiceStatus object with a status code 200.

pyninja.features.service.kickstart_mac_service(full_service_name: str) → ServiceStatus

Kickstart a macOS service by name.

Parameters:

full_service_name – Full name of the service, including the prefix.

References

A service can be restarted using the launchctl kickstart command on macOS.

launchctl kickstart -k gui/$(id -u)/com.ollama.ollama
Returns:

Returns an instance of the ServiceStatus object.

Return type:

ServiceStatus

pyninja.features.service.restart_service(service_name: str) → ServiceStatus

Restart a service by name.

Parameters:

service_name – Name of the service.

Returns:

Returns an instance of the ServiceStatus object.

Return type:

ServiceStatus

Zipper

pyninja.features.zipper.archive(path: Path, directory: Path) → Path

Archives a file or directory into a zip file.

Parameters:

  • path – Path to the file or directory to be archived.

  • directory – Directory where the zip file will be created.

Returns:

The path to the created zip file.

Return type:

pathlib.Path

pyninja.features.zipper.unarchive(zip_file: str | pathlib.Path, directory: str | pathlib.Path) → str

Unarchives a zip file into a specified directory.

Parameters:

  • zip_file – Zip file to be unarchived.

  • directory – Directory where the contents will be extracted.

Returns:

The path to the directory containing the unzipped content.

Return type:

str

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 for both sync and async functions.

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

  • Works with both synchronous and asynchronous functions.

  • For async functions, caches the awaited result, not the coroutine itself.

  • Custom async cache implementation to handle coroutine lifecycle properly.

Notes

  • 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_large_file: str = '/get-large-file'
put_large_file: str = '/put-large-file'
delete_content: str = '/delete-content'
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'
restart_service: str = '/restart-service'
get_all_apps: str = '/get-all-apps'
start_app: str = '/start-app'
stop_app: str = '/stop-app'
restart_app: str = '/restart-app'
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'
get_mfa: str = '/get-mfa'
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
current_dir: str
swagger_ui: str
mfa_template: 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
gmail_user: pydantic.networks.EmailStr | None
gmail_pass: str | None
recipient: str | None
mfa_timeout: int
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
osascript: Path
mdls: Path
open: 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.keygen(min_length: int = 32) → str

Key generator to create a unique key with a minimum length.

Parameters:

min_length – Minimum length of the generated key. Default is 32.

Returns:

Returns a unique key that contains at least one digit, one uppercase letter,

Return type:

str

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.AppStatus(*, app_name: str, status_code: int, description: str, app_path: str | None = None)

Object to load the application status with a status code and description.

>>> AppStatus
app_name: str
status_code: int
description: str
app_path: str | None
_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]] = {'app_name': FieldInfo(annotation=str, required=True), 'app_path': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'description': FieldInfo(annotation=str, required=True), 'status_code': FieldInfo(annotation=int, required=True)}

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.MFAToken(*, token: str | None = None, timers: List[Timer] = None)

Object to store the MFA token.

>>> MFAToken
token: str | None
timers: List[Timer]
class Config

Configuration for MFAToken object.

arbitrary_types_allowed = True
_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] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'timers': FieldInfo(annotation=List[Timer], required=False, default_factory=list), 'token': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

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.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(current_dir: str) → str

Get the custom JavaScript for Swagger UI.

pyninja.modules.models.load_mfa_template(current_dir: str) → str

Get the custom HTML template for MFA template.

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.

Tree - Directory Structure

class pyninja.modules.tree.Tree(skip_dot_files: bool)

Root tree formatter for a particular directory location.

This class allows the creation of a visual representation of the file system hierarchy of a specified directory. It can optionally skip files that start with a dot (hidden files).

>>> Tree
scan(path: Path, last: bool = True, header: str = '') → List[str]

Returns contents for a folder as a root tree.

Parameters:

  • path – Directory path for which the root tree is to be extracted.

  • last – Indicates if the current item is the last in the directory.

  • header – The prefix for the current level in the tree structure.

Returns:

A list of strings representing the directory structure.

Return type:

List[str]

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]

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:

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]

async pyninja.monitor.resources.get_disk_info() → List[Dict[str, int | str]]

Get partition and usage information for each physical drive.

Returns:

Returns a list of key-value pairs with ID, name, and usage.

Return type:

List[Dict[str, str | int]]

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

Get CPU cores configured for a particular container using NanoCpus.

Parameters:

container_id – ID of the Docker container to inspect.

Returns:

Returns the number of CPU cores.

Return type:

int

pyninja.monitor.resources.floater(value: float) → int | float

Convert a float to an int if it is a whole number, otherwise return the float.

Parameters:

value – Value to convert.

Returns:

Returns an int if the value is a whole number, otherwise returns the float.

Return type:

int | float

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 | None

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.

Indices and tables