Welcome to PyNinja’s documentation!¶

Contents:

PyNinja

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.
- host_password - Password to authenticate sudo commands on the server.
- observability_session - Session timeout for observability metrics.
Multifactor_Authentication –
- mfa_timeout - Timeout duration for MFA in seconds.
- mfa_resend_delay - Resend duration for MFA in seconds.
- Email
  - gmail_user - Gmail username for MFA.
  - gmail_pass - Gmail password for MFA.
  - recipient - Recipient email address for MFA. Defaults to gmail_user
- Ntfy
  - ntfy_url - Ntfy server URL.
  - ntfy_topic - Subscribed ntfy topic.
- Telegram
  - telegram_token - Telegram bot token.
  - telegram_chat_id - Telegram chat ID to send MFA.
- Authenticator
  - authenticator_app - Authenticator app name.
  - authenticator_user - Authenticator username.
  - authenticator_token - MFA Authenticator token.
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.
- 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.
Certificates –
- certbot_path - Path to the certbot binary for certificate management.
- cert_scan - Dictionary of key-value pairs threshold and schedule.
  - threshold - Expiry threshold in days to trigger alerts.
  - schedule - Time in %H:%M format to run the certificate monitoring job daily.
macOS_Specific_Binaries –
- osascript - Path to the osascript binary for macOS.
- mdls - Path to the mdls binary for macOS.
- open - Path to the open binary for macOS.
Windows_Specific_Binaries –
- pwsh - Path to the pwsh binary for Windows.

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: APIRoute) → 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) → starlette.responses.JSONResponse | starlette.responses.RedirectResponse¶

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: fastapi.security.http.HTTPAuthorizationCredentials | None) → 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.validate_otp(code: str) → bool¶

Validate MFA generated through authenticator.

Parameters:: code – OTP code generated from authenticator app.
Returns:: Returns a boolean flag to indicate if the OTP code is valid.
Return type:: bool

pyninja.executors.auth.verify_mfa(mfa_code: str) → bool¶

Verifies the multifactor authentication code.

Parameters:: mfa_code – Multifactor authentication code to verify.
Returns:: Returns a boolean flag to indicate if the MFA code is valid.
Return type:: bool

async pyninja.executors.auth.level_2(request: Request, apikey: fastapi.security.http.HTTPAuthorizationCredentials | None, api_secret: str | None, mfa_code: str | None) → 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_token(table: TableName, get_all: bool = False) → Any | None¶

Gets run/mfa token stored in the database.

Returns:: Returns the token stored in the database.
Return type:: Any

pyninja.executors.database.update_token(table: TableName, token: str = None, requester: MFAOptions = None, expiry: int = 0) → None¶

Update run/mfa token in the database.

Parameters:

table – Table name to update the token.
token – Token to be stored in the database.
requester – MFA type.
expiry – Expiry time in seconds from the current epoch time.

pyninja.executors.database.get_forbidden(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_forbidden(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_forbidden(host: str) → None¶

Deletes all records related to the host address.

Parameters:: host – Host address.

API Routes¶

async pyninja.routes.commands.run_ui(request: Request)¶: Renders the HTML template for the run command UI.

async pyninja.routes.commands.run_command(request: ~starlette.requests.Request, payload: ~pyninja.modules.payloads.RunCommand, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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.

pyninja.routes.fullaccess.create_directory(directory: Union[Path, str]) → 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.delete_content(request: ~starlette.requests.Request, payload: ~pyninja.modules.payloads.DeleteContent, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, payload: ~pyninja.modules.payloads.ListFiles, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, payload: ~pyninja.modules.payloads.GetFile, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.Optional[str] = Header(None))¶

Download a file from the server.

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: ~starlette.requests.Request, file: ~fastapi.datastructures.UploadFile, directory: ~pathlib.Path, overwrite: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.Optional[str] = Header(None))¶

Upload a file to the server.

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: ~starlette.requests.Request, public: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, interval: int | float = 2, per_cpu: bool = True, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, path: str = '/', apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, process_name: str, cpu_interval: ~typing.Union[int, float] = 1, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, service_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, process_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, service_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, service_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, service_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, service_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, app_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, app_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, app_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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.observability.get_observability(request: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), interval: int = 3, session_duration: int = 3600, all_services: bool = False)¶

API function to get system metrics via StreamingResponse.

Args:

request: Reference to the FastAPI request object.

apikey: API Key to authenticate the request.

interval: Sleep interval for streaming.

session_duration: Number of seconds for the observability session.

all_services: Flag to include all services in the services’ metrics.

Raises:

StreamingResponse: Streams system resources information.

async pyninja.routes.orchestration.get_docker_containers(request: ~starlette.requests.Request, container_name: str = None, get_all: bool = False, get_running: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, container_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, container_name: str, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

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

async pyninja.executors.routers.version()¶

Get PyNinja and Python version information.

Returns:: Returns the version information as key-value pairs.
Return type:: dict

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.YIELD_LINE(inline)¶

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

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

async pyninja.executors.squire.stream_command(request: Request, command: str, shell: bool, timeout: int | float, stream_limit: int | float) → AsyncGenerator[str, None]¶

Async generator to stream command output line-by-line and handle disconnects.

Parameters:

request – FastAPI request object.
command – Command to be executed.
shell – Boolean flag to indicate shell execution.
timeout – Timeout for the command execution.
stream_limit – Time in seconds to stream.

Yields:

str – Lines of output from the command execution.

pyninja.executors.squire.process_command(command: str, timeout: 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() → Architecture¶

Load architecture details from environment variables.

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

pyninja.executors.squire.generate_mfa_token(length: int = 32) → str¶

Key generator to create a unique key with an exact length.

Parameters:: length – Exact length of the generated key. Default is 32.
Returns:: Returns a unique key that contains at least one digit, one uppercase letter, one lowercase letter, and one safe character.
Return type:: str

pyninja.executors.squire.keygen(nbytes: int = 64) → str¶

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

Parameters:: nbytes – Number of random bytes (nbytes) to generate.
Returns:: Returns a URL safe token.
Return type:: str

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.assert_cert_monitor() → None¶: Ensure at least one notification method is configured for cert_monitor if it is enabled.

pyninja.executors.squire.any_mfa_enabled() → bool¶

Check if any MFA method is enabled.

Returns:: Returns True if any MFA method is enabled, else False.
Return type:: bool

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 | int | float]¶: Convert the usage metrics into human-readable format.

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

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

pyninja.executors.squire.mask_sensitive_data(text: str) → str¶

Mask sensitive data in the given text.

Parameters:: text – Text to mask sensitive data from.
Returns:: Returns the text with sensitive data masked.
Return type:: str

pyninja.executors.squire.log_subprocess_error(error: CalledProcessError) → str¶

Log subprocess error with return code, error reason, and command.

Parameters:: error – Subprocess error object.
Returns:: Returns the error reason as a string.
Return type:: str

pyninja.executors.squire.get_mfa_title(include_node: bool) → str¶

Generate MFA title with current date and time.

Parameters:: include_node – Boolean flag to include node name in the title.
Returns:: Returns the MFA title.
Return type:: str

class pyninja.executors.squire.AddProcessName(process_name: str)¶

Wrapper that overrides logging.Filter to add processName to the existing log format.

>>> AddProcessName

Parameters:: process_name – Takes name of the process to be added as argument.

filter(record: LogRecord) → bool¶: Overrides the built-in filter record.

class pyninja.executors.squire.HealthCheckFilter(name='')¶

Custom logging filter to exclude health check logs from the output.

>>> HealthCheckFilter

filter(record: LogRecord) → bool¶: Filter out logs related to health checks.

Multifactor Authentication¶

Authenticator¶

This module provides functionality for generating a QR code for TOTP (Time-based One-Time Password) setup.

Note

To generate a QR code, run one of the following:

Option 1: Python snippet

import pyninja
pyninja.otp.generate_qr(show_qr=True)

Option 2: CLI command

pyninja --mfa

pyninja.multifactor.otp.display_secret(secret: str, qr_filename: str) → None¶

Displays the TOTP secret key.

Parameters:

secret – The TOTP secret key to display.
qr_filename – The filename where the QR code is saved.

pyninja.multifactor.otp.generate_qr(show_qr: bool = True) → None¶

Generates a QR code for TOTP setup.

Parameters:: - show_qr – If True, displays the QR code using the default image viewer.

Initializer¶

async pyninja.multifactor.mfa.send_new_mfa() → Optional[NoReturn]¶: Function to check if a new MFA token should be sent.

async pyninja.multifactor.mfa.get_mfa(request: ~starlette.requests.Request, mfa_option: ~pyninja.modules.enums.MFAOptions, get_node: bool = True, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

Get multifactor authentication code.

Args:

request: Reference to the FastAPI request object.

get_node: Boolean flag to include node name in the title.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code.

async pyninja.multifactor.mfa.delete_mfa(request: ~starlette.requests.Request, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

Delete 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.

Gmail¶

pyninja.multifactor.gmail.instantiate_mailer() → gmailconnector.send_email.SendEmail | None¶

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

async pyninja.multifactor.gmail.send(subject: str, body: str = None, html_body: str = None) → bool¶

Function to send an email using the gmail-connector module.

Parameters:

subject – Subject of the email.
body – Body of the email. Optional if html_body is provided.
html_body – HTML body of the email. Optional if body is provided.

Returns:

Returns True if the email was sent successfully, False otherwise.

Return type:

bool

async pyninja.multifactor.gmail.get_mfa(request: ~starlette.requests.Request, get_node: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

Get multifactor authentication code.

Args:

request: Reference to the FastAPI request object.

get_node: Boolean flag to include node name in the title.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code to indicate MFA delivery.

Ntfy¶

async pyninja.multifactor.ntfy.send(title: str, data: str) → bool¶

Function to send a notification via Ntfy.

Parameters:

title – Ntfy notification title.
data – Data to be sent in the notification body.

Returns:

Returns True if the notification was sent successfully, False otherwise.

Return type:

bool

async pyninja.multifactor.ntfy.get_mfa(request: ~starlette.requests.Request, get_node: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

Get multifactor authentication code via Ntfy.

Args:

request: Reference to the FastAPI request object.

get_node: Boolean flag to include node name in the title.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code to indicate MFA delivery.

Telegram¶

async pyninja.multifactor.telegram.send(message: str, parse_mode: str | None = 'markdown') → bool¶

Sends a message to the user via Telegram.

Parameters:

message – Message to be sent to the user.
parse_mode – Parse mode. Defaults to markdown

Returns:

Returns True if the message was sent successfully, False otherwise.

Return type:

bool

async pyninja.multifactor.telegram.get_mfa(request: ~starlette.requests.Request, get_node: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

Get multifactor authentication code via Telegram.

Args:

request: Reference to the FastAPI request object.

get_node: Boolean flag to include node name in the title.

apikey: API Key to authenticate the request.

Raises:

APIResponse: Raises the HTTPStatus object with a status code to indicate MFA delivery.

API Routes - Certificates¶

Certificates¶

Module to handle certificates API endpoint.

async pyninja.routes.certificates.get_certificate(request: ~starlette.requests.Request, name: ~typing.Optional[str] = 'all', apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None))¶

API handler to get certificates.

Parameters:

- request – Reference to the FastAPI request object.
- name – Name of the certificate to retrieve.
- raw – If True, returns raw certificate data instead of parsed model.
- apikey – API Key to authenticate the request.

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: ~starlette.requests.Request, filepath: ~typing.Optional[~pathlib.Path] = None, directory: ~typing.Optional[~pathlib.Path] = None, chunk_size: int = 8192, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: ~starlette.requests.Request, filename: str, directory: str, part_number: int = 0, is_last: bool = False, checksum: ~typing.Optional[str] = None, overwrite: bool = False, unzip: bool = False, delete_after_unzip: bool = False, apikey: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None), api_secret: ~typing.Optional[str] = Header(None), mfa_code: ~typing.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: str, 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

Cert Expiration¶

async pyninja.features.cert_expiration.scheduler() → None¶: Schedule the certificate expiry check to run daily at a specified time.

pyninja.features.cert_expiration.html_body(rows: List[Tuple[str, str, str, str]]) → str¶

Construct the HTML body for the certificate expiry report email.

Parameters:: rows – A list of tuples containing certificate details (name, expiry date, status, message).
Returns:: The HTML body for the certificate expiry report email.
Return type:: str

async pyninja.features.cert_expiration.notifier(rows: List[Tuple[str, str, str, str]]) → None¶

Notify the user about expiring certificates.

Parameters:: rows – A list of tuples containing certificate details (name, expiry date, status, message).

async pyninja.features.cert_expiration.monitor_expiry() → None¶: Check for certificates expiring within 10 days and log a warning if any are found.

Certificates¶

pyninja.features.certificates.forbidden() → CertificateStatus¶

Return a forbidden response when host password is not stored.

Returns:: A CertificateStatus object indicating that the host password is not stored.
Return type:: models.CertificateStatus

pyninja.features.certificates.parse_certificate_output(output: str, raw: bool = False, ws_stream: bool = False) → Generator[Dict[str, Any]]¶

Parse the output from the certbot command to extract certificate details.

Parameters:

output – The output string from the certbot command.
raw – If True, returns raw certificate data instead of parsed model.
ws_stream – If True, omits file paths and serial numbers for simplicity.

Yields:

Dict[str, Any] – A dictionary representing one certificate.

async pyninja.features.certificates.get_all_certificates(raw: bool = False, ws_stream: bool = False) → CertificateStatus¶

Fetch all SSL certificates using certbot.

Returns:: A CertificateStatus object containing the status code and a list of certificates.
Return type:: models.CertificateStatus

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) → str | None¶

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) → str | None¶

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) → Dict[str, int | 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]

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

Get parent processes from a list of processes.

Parameters:: processes – List of process dictionaries.
Returns:: Returns a list of dictionaries with unique parent processes.
Return type:: List[Dict[str, str]]

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: Union[int, float]) → List[Dict[str, None]]¶

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, Any]

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

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, Any]

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

Constructs an ServiceStatus object with a status code 403.

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, directory: str) → 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 = 'linux'¶

darwin = 'darwin'¶

windows = '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 = '/'¶

docs = '/docs'¶

redoc = '/redoc'¶

health = '/health'¶

version = '/version'¶

get_file = '/get-file'¶

put_file = '/put-file'¶

get_large_file = '/get-large-file'¶

put_large_file = '/put-large-file'¶

delete_content = '/delete-content'¶

get_processor = '/get-processor'¶

get_all_services = '/get-all-services'¶

get_service_status = '/get-service-status'¶

get_service_usage = '/get-service-usage'¶

stop_service = '/stop-service'¶

start_service = '/start-service'¶

restart_service = '/restart-service'¶

get_all_apps = '/get-all-apps'¶

start_app = '/start-app'¶

stop_app = '/stop-app'¶

restart_app = '/restart-app'¶

get_process_status = '/get-process-status'¶

get_process_usage = '/get-process-usage'¶

observability = '/observability'¶

monitor = '/monitor'¶

ws_system = '/ws/system'¶

run_command = '/run-command'¶

run_ui = '/run-ui'¶

get_memory = '/get-memory'¶

get_all_disks = '/get-all-disks'¶

get_disk_utilization = '/get-disk-utilization'¶

get_docker_images = '/get-docker-images'¶

get_docker_stats = '/get-docker-stats'¶

get_docker_volumes = '/get-docker-volumes'¶

get_docker_containers = '/get-docker-containers'¶

get_certificates = '/get-certificates'¶

renew_certificate = '/renew-certificate'¶

stop_docker_container = '/stop-docker-container'¶

start_docker_container = '/start-docker-container'¶

get_ip = '/get-ip'¶

get_cpu = '/get-cpu'¶

list_files = '/list-files'¶

get_cpu_load = '/get-cpu-load'¶

login = '/login'¶

logout = '/logout'¶

error = '/error'¶

get_mfa = '/get-mfa'¶

delete_mfa = '/delete-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 = 'drive'¶

monitor = 'monitor'¶

session_token = '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 = 'main.html'¶

index = 'index.html'¶

logout = 'logout.html'¶

session = 'session.html'¶

unauthorized = 'unauthorized.html'¶

disk_report_linux = 'disk_report_linux.html'¶

disk_report_darwin = 'disk_report_darwin.html'¶

run_ui = 'run_ui.html'¶

swagger_ui = 'swaggerUI.js'¶

mfa_template = 'mfa_template.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 = 'get'¶

post = 'post'¶

monitor = 'monitor'¶

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

Table names.

>>> TableName

auth_errors = 'auth_errors'¶

mfa_token = 'mfa_token'¶

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

Authentication mechanism for multifactor tokens.

>>> MFAOptions

email = 'email'¶

ntfy = 'ntfy'¶

telegram = 'telegram'¶

Exceptions¶

exception pyninja.modules.exceptions.APIResponse(status_code: int, detail: Any = None, headers: collections.abc.Mapping[str, str] | None = 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

forbid: Set[str]¶

rps: Dict[str, int]¶

auth_counter: Dict[str, int]¶

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¶

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¶

host_password: str | None¶

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

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

remote_execution: bool¶

api_secret: str | None¶

database: str¶

authenticator_app: str¶

authenticator_user: str | pydantic.networks.EmailStr¶

authenticator_token: str | None¶

gmail_user: pydantic.networks.EmailStr | None¶

gmail_pass: str | None¶

recipient: str | None¶

ntfy_url: pydantic.networks.HttpUrl | None¶

ntfy_username: str | None¶

ntfy_password: str | None¶

ntfy_topic: str | None¶

telegram_token: str | None¶

telegram_chat_id: str | None¶

mfa_resend_delay: int¶

mfa_timeout: int¶

monitor_username: str | None¶

monitor_password: str | None¶

monitor_session: int¶

disk_report: bool¶

max_connections: int¶

processes: List[str]¶

services: List[str]¶

smart_lib: Optional[Path]¶

gpu_lib: Path¶

disk_lib: Path¶

service_lib: Path¶

processor_lib: Path¶

docker_lib: Optional[Path]¶

certbot_path: Optional[Path]¶

cert_scan: pyninja.modules.models.CertScan | None¶

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

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.CertificateStatus(*, status_code: int, description: str, certificates: ~typing.Union[~typing.List[~typing.Dict[str, ~typing.Any]], ~typing.List[str]] = <factory>)¶

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

>>> ServiceStatus

status_code: int¶

description: str¶

certificates: Union[List[Dict[str, Any]], List[str]]¶

_abc_impl = <_abc._abc_data object>¶

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

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_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pyninja.modules.models.default_service_lib() → Dict[str, str]¶

Get default service library filepath for the host operating system.

Returns:: Returns a mapping of operating systems to their default service library filepaths.
Return type:: Dict[str, str]

pyninja.modules.models.default_docker_lib() → Dict[str, str]¶

Get the default docker library filepath for the host operating system.

Returns:: Returns a mapping of operating systems to their default docker library filepaths.
Return type:: Dict[str, str]

pyninja.modules.models.get_optional_lib(func: Callable) → Optional[Path]¶

Get the optional library filepath for the host operating system.

Parameters:: func – Function to call to get the mapping.
Returns:: Returns the filepath as a FilePath object if it exists, otherwise None.
Return type:: FilePath | None

pyninja.modules.models.retrieve_library_path(func: Callable, ignore_err: bool = False) → Path¶

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

Parameters:

func – Function to call to get the mapping.
ignore_err – If True, ignores the KeyError and returns an empty FilePath.

Returns:

Returns the library path as a FilePath object.

Return type:

FilePath

pyninja.modules.models.get_certbot_path() → Optional[Path]¶

Get the certbot path if installed.

Returns:: Returns the FilePath referencing the certbot binary.
Return type:: FilePath

class pyninja.modules.models.CertScan(*, threshold: int = 10, schedule: str = '8:00')¶

Object to load the certificate scan settings.

>>> CertScan

threshold: int¶

schedule: str¶

class Config¶

Configuration for certificate scan settings.

extra = 'forbid'¶

_abc_impl = <_abc._abc_data object>¶

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

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

pyninja.modules.models.load_mfa_template(source: str) → str¶: Get the custom HTML template for MFA template.

class pyninja.modules.models.DatabaseConnection(datastore: str, timeout: int)¶

Context manager for SQLite database connections.

>>> DatabaseConnection

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]], drop_existing: bool = False) → 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.
drop_existing – If True, drops the existing table before creating a new one.

describe_database() → Dict[str, Any]¶

Returns a summary of each table in the database, including column names and record counts.

Returns:: A dictionary where keys are table names and values are dicts with column names and record counts.
Return type:: Dict[str, Any]

Payloads¶

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

Payload for run-command endpoint.

>>> RunCommand

command: str¶

shell: bool¶

timeout: Union[int, float]¶

stream: bool¶

stream_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) → Union[Dict[str, int], NoReturn]¶

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

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]

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, Optional[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, Optional[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[Any], float]]¶

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:: Dict[str, str | List[Any] | float]

async pyninja.monitor.resources.service_obs() → List[Dict[str, str]]¶: Awaits the result of all services before returning.

async pyninja.monitor.resources.system_resources(all_services: bool = False) → Dict[str, Union[dict, List[Dict[str, int | str]]]]¶

Gather system resources including Docker stats asynchronously.

Parameters:: all_services – Boolean flag to get all services.
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: ~starlette.requests.Request, authorization: ~fastapi.security.http.HTTPAuthorizationCredentials = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None)) → 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¶

Multifactor Authentication¶

Authenticator¶

Initializer¶

Gmail¶

Ntfy¶

Telegram¶

API Routes - Certificates¶

Certificates¶

API Routes - Large Scale¶

Download¶

Upload¶

PyNinja - Features¶

Application¶

Cert Expiration¶

Certificates¶

Docker¶

Operations¶

Process¶

Service¶

Zipper¶

PyNinja - Modules¶

Cache¶

Enums¶

Exceptions¶

Models¶

Payloads¶

RateLimit¶

Secure¶

Tree - Directory Structure¶

PyNinja - Monitor¶

Authenticator¶

Configuration¶

Drive¶

Resources¶

Routes¶

Indices and tables¶