Welcome to PyNinja’s documentation!¶
PyNinja - Main¶
- pyninja.main.get_desc(remote_flag: bool, monitor_flag: bool) str ¶
Construct a detailed description for the API docs.
- Parameters:
remote_flag – Boolean flag to indicate remote execution state.
monitor_flag – Boolean flag to indicate monitoring page state.
- Returns:
Returns the description as a string.
- Return type:
str
- async pyninja.main.redirect_exception_handler(request: Request, exception: RedirectException) JSONResponse ¶
Custom exception handler to handle redirect.
- Parameters:
request – Takes the
Request
object as an argument.exception – Takes the
RedirectException
object inherited fromException
as an argument.
- Returns:
Returns the JSONResponse with content, status code and cookie.
- Return type:
JSONResponse
- pyninja.main.start(**kwargs) None ¶
Starter function for the API, which uses uvicorn server as trigger.
- Keyword Arguments:
env_file – Env filepath to load the environment variables.
apikey – API Key for authentication.
ninja_host – Hostname for the API server.
ninja_port – Port number for the API server.
remote_execution – Boolean flag to enable remote execution.
api_secret – Secret access key for running commands on server remotely.
monitor_username – Username to authenticate the monitoring page.
monitor_password – Password to authenticate the monitoring page.
monitor_session – Session timeout for the monitoring page.
service_manager – Service manager filepath to handle the service status requests.
database – FilePath to store the auth database that handles the authentication errors.
rate_limit – List of dictionaries with
max_requests
andseconds
to apply as rate limit.log_config – Logging configuration as a dict or a FilePath. Supports .yaml/.yml, .json or .ini formats.
PyNinja - Executors¶
API Authenticator¶
- pyninja.executors.auth.EPOCH()¶
- async pyninja.executors.auth.forbidden(request: Request) None ¶
Validates if a request is part of the forbidden list.
- Parameters:
request – Reference to the FastAPI request object.
- Raises:
- 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:
- 401 – If authorization is invalid.
- 403 – If host address is forbidden.
- async pyninja.executors.auth.level_2(request: Request, apikey: HTTPAuthorizationCredentials, token: str) None ¶
Validates the auth request using HTTPBearer and additionally a secure token.
- Parameters:
request – Takes the authorization header token as an argument.
apikey – Basic APIKey required for all the routes.
token – Additional token for critical requests.
- Raises:
- 401 – If authorization is invalid.
- 403 – If host address is forbidden.
- async pyninja.executors.auth.incrementer(attempt: int) int ¶
Increments block time for a host address based on the number of failed login attempts.
- Parameters:
attempt – Number of failed login attempts.
- Returns:
Returns the appropriate block time in minutes.
- Return type:
int
- async pyninja.executors.auth.handle_auth_error(request: Request) None ¶
Handle authentication errors from the filebrowser API.
- Parameters:
request – The incoming request object.
Database¶
- pyninja.executors.database.get_record(host: str) int | None ¶
Gets blocked epoch time for a particular host.
- Parameters:
host – Host address.
- Returns:
Returns the epoch time until when the host address should be blocked.
- Return type:
int
- pyninja.executors.database.put_record(host: str, block_until: int) None ¶
Inserts blocked epoch time for a particular host.
- Parameters:
host – Host address.
block_until – Epoch time until when the host address should be blocked.
- pyninja.executors.database.remove_record(host: str) None ¶
Deletes all records related to the host address.
- Parameters:
host – Host address.
API Routes¶
- async pyninja.routes.fullaccess.run_command(request: Request, payload: RunCommand, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶
API function to run a command on host machine.
Args:
request: Reference to the FastAPI request object.
payload: Payload received as request body.
apikey: API Key to authenticate the request.
token: API secret to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.fullaccess.list_files(request: Request, payload: ListFiles, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶
Get all YAML files from fileio and all log files from logs directory.
Args:
request: Reference to the FastAPI request object.
payload: Payload received as request body.
apikey: API Key to authenticate the request.
token: API secret to authenticate the request.
Returns:
Dict[str, List[str]]: Dictionary of files that can be downloaded or uploaded.
- async pyninja.routes.fullaccess.get_file(request: Request, payload: GetFile, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶
Download a particular YAML file from fileio or log file from logs directory.
Args:
request: Reference to the FastAPI request object.
payload: Payload received as request body.
apikey: API Key to authenticate the request.
token: API secret to authenticate the request.
Returns:
FileResponse: Returns the FileResponse object of the file.
- async pyninja.routes.fullaccess.put_file(request: Request, file: UploadFile, directory: Path, overwrite: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer), token: Optional[str] = Header(None))¶
Upload a file to th.
Args:
request: Reference to the FastAPI request object.
file: Upload object for the file param.
payload: Payload received as request body.
apikey: API Key to authenticate the request.
token: API secret to authenticate the request.
- async pyninja.routes.ipaddr.get_ip_address(request: Request, public: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get local and public IP address of the device.
Args:
request: Reference to the FastAPI request object.
public: Boolean flag to get the public IP address.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and the public/private IP as response.
- async pyninja.routes.metrics.get_cpu_utilization(request: Request, interval: int | float = 2, per_cpu: bool = True, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get the CPU utilization.
Args:
request: Reference to the FastAPI request object.
interval: Interval to get the CPU utilization.
per_cpu: If True, returns the CPU utilization for each CPU.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.
- async pyninja.routes.metrics.get_memory_utilization(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get memory utilization.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.
- async pyninja.routes.metrics.get_cpu_load_avg(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get the number of processes in the system run queue averaged over the last 1, 5, and 15 minutes respectively.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.
- async pyninja.routes.metrics.get_disk_utilization(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get disk utilization.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and CPU usage as response.
- async pyninja.routes.metrics.get_all_disks(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get all disks attached to the host device.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and attached disks as response.
- async pyninja.routes.namespace.get_process_status(request: Request, process_name: str, cpu_interval: Union[int, float] = 1, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to monitor a process.
Args:
request: Reference to the FastAPI request object.
process_name: Name of the process to check status.
cpu_interval: Interval in seconds to get the CPU usage.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.namespace.get_service_usage(request: Request, service_names: List[str], apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to monitor a service.
Args:
request: Reference to the FastAPI request object.
service_names: Name of the service to check status.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.namespace.get_process_usage(request: Request, process_names: List[str], apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to monitor a process.
Args:
request: Reference to the FastAPI request object.
process_names: Name of the service to check status.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.namespace.get_service_status(request: Request, service_name: str, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to get the status of a service.
Args:
request: Reference to the FastAPI request object.
service_name: Name of the service to check status.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.namespace.get_processor_name(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to get process information.
Args:
request: Reference to the FastAPI request object.
process_name: Name of the process to get information.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.orchestration.get_docker_containers(request: Request, container_name: str = None, get_all: bool = False, get_running: bool = False, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to get docker containers’ information.
Args:
request: Reference to the FastAPI request object.
container_name: Name of the container to check status.
get_all: Get all the containers’ information.
get_running: Get running containers’ information.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.orchestration.get_docker_images(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to get docker images’ information.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.orchestration.get_docker_volumes(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
API function to get docker volumes’ information.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and detail as response.
- async pyninja.routes.orchestration.get_docker_stats(request: Request, apikey: HTTPAuthorizationCredentials = Depends(HTTPBearer))¶
Get docker-stats for all running containers.
Args:
request: Reference to the FastAPI request object.
apikey: API Key to authenticate the request.
Raises:
APIResponse: Raises the HTTPStatus object with a status code and attached disks as response.
- async pyninja.executors.routers.docs() RedirectResponse ¶
Redirect to docs page.
- Returns:
Redirects the user to
/docs
page.- Return type:
RedirectResponse
- async pyninja.executors.routers.health()¶
Health check for PyNinja.
- Returns:
Returns a health check response with status code 200.
- Return type:
- pyninja.executors.routers.get_all_routes(dependencies: List[Depends]) List[APIRoute] ¶
Get all the routes to be added for the API server.
- Parameters:
dependencies – List of dependencies to be added to the routes
- Returns:
Returns the routes as a list of APIRoute objects.
- Return type:
List[APIRoute]
Squire¶
- pyninja.executors.squire.public_ip_address() str ¶
Gets public IP address of the host using different endpoints.
- Returns:
Public IP address.
- Return type:
str
- pyninja.executors.squire.private_ip_address() str | None ¶
Uses a simple check on network id to see if it is connected to local host or not.
- Returns:
Private IP address of host machine.
- Return type:
str
- pyninja.executors.squire.format_nos(input_: float) int | float ¶
Removes
.0
float values.- Parameters:
input_ – Strings or integers with
.0
at the end.- Returns:
Int if found, else returns the received float value.
- Return type:
int | float
- pyninja.executors.squire.format_timedelta(td: timedelta) str ¶
Converts timedelta to human-readable format by constructing a formatted string based on non-zero values.
- Parameters:
td – Timedelta object.
See also
Always limits the output to a maximum of two identifiers.
Examples
3 days and 1 hour
1 hour and 11 minutes
5 minutes and 23 seconds
- Returns:
Human-readable format of timedelta.
- Return type:
str
- pyninja.executors.squire.size_converter(byte_size: int | float) str ¶
Gets the current memory consumed and converts it to human friendly format.
- Parameters:
byte_size – Receives byte size as argument.
- Returns:
Converted understandable size.
- Return type:
str
- pyninja.executors.squire.process_command(command: str, timeout: Union[int, float]) Dict[str, List[str]] ¶
Process the requested command.
- Parameters:
command – Command as string.
timeout – Timeout for the command.
- Returns:
Returns the result with stdout and stderr as key-value pairs.
- Return type:
Dict[str, List[str]]
- pyninja.executors.squire.envfile_loader(filename: str | os.PathLike) EnvConfig ¶
Loads environment variables based on filetypes.
- Parameters:
filename – Filename from where env vars have to be loaded.
- Returns:
Returns a reference to the
EnvConfig
object.- Return type:
- 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:
- pyninja.executors.squire.keygen() str ¶
Generate session token from secrets module, so that users are forced to log in when the server restarts.
- Returns:
Returns a URL safe 64-bit token.
- Return type:
str
- pyninja.executors.squire.dynamic_numbers(string: str) int | float | None ¶
Convert strings to integer or float dynamically.
- Parameters:
string – Number in string format.
- Returns:
Integer or float value.
- Return type:
int | float
PyNinja - Features¶
CPU¶
- pyninja.features.cpu._darwin(lib_path: Path) str ¶
Get processor information for macOS.
- Parameters:
lib_path – Path to the library executable.
- pyninja.features.cpu._linux(lib_path: Path) str ¶
Get processor information for Linux.
- Parameters:
lib_path – Path to the library file.
- pyninja.features.cpu._windows(lib_path: Path) str ¶
Get processor information for Windows.
- Parameters:
lib_path – Path to the library file.
- pyninja.features.cpu.get_name() str | None ¶
Get processor information for the host operating system.
- Returns:
Returns the processor information as a string.
- Return type:
str
Disks¶
- pyninja.features.disks.get_partitions_for_disk(device_id: str) List[str] ¶
Use psutil to find partitions for a given device.
- Parameters:
device_id
- Returns:
Returns the list of partitions in a disk.
- Return type:
List[str]
- pyninja.features.disks.parse_size(size_str) str ¶
Convert size string with units to a standard size in bytes.
- Parameters:
size_str – Size with unit as string.
- Returns:
Returns the parsed size as string with units attached.
- Return type:
str
- pyninja.features.disks._linux(lib_path: Path) List[Dict[str, str]] ¶
Get disks attached to Linux devices.
- Parameters:
lib_path – Returns the library path for disk information.
- Returns:
Returns disks information for Linux distros.
- Return type:
List[Dict[str, str]]
- pyninja.features.disks.is_physical_disk(lib_path: Path, device_id: str) bool ¶
Check if the disk is a physical disk using diskutil info.
- Parameters:
lib_path – Library path to get disk info.
device_id – Device mount ID.
- Returns:
Boolean to indicate virtual or physical
- Return type:
bool
- pyninja.features.disks._darwin(lib_path: Path) List[Dict[str, str]] ¶
Get disks attached to macOS devices.
- Parameters:
lib_path – Returns the library path for disk information.
- Returns:
Returns disks information for macOS devices.
- Return type:
List[Dict[str, str]]
- pyninja.features.disks._reformat_windows(data: Dict[str, str | int | float]) Dict[str, str] ¶
Reformats each drive’s information for Windows OS.
- Parameters:
data – Data as a dictionary.
- Returns:
Returns a dictionary of key-value pairs.
- Return type:
Dict[str, str]
- pyninja.features.disks._windows(lib_path: Path) List[Dict[str, str]] ¶
Get disks attached to Windows devices.
- Parameters:
lib_path – Returns the library path for disk information.
- Returns:
Returns disks information for Windows machines.
- Return type:
List[Dict[str, str]]
- pyninja.features.disks.get_all_disks() List[Dict[str, str]] ¶
OS-agnostic function to get all disks connected to the host system.
Docker¶
- pyninja.features.dockerized.get_container_status(name: str = None) str | None ¶
Get container status by name.
- Parameters:
name – Name of the container or image used to run the container.
- Returns:
Container status as a string.
- Return type:
str
- pyninja.features.dockerized.get_running_containers() Generator[Dict[str, str]] ¶
Get running containers.
- Yields:
Dict[str, str] – Yields a dictionary of running containers with the corresponding metrics.
- pyninja.features.dockerized.get_all_containers() Optional[List[Dict[str, str]]] ¶
Get all containers and their metrics.
- Returns:
Returns a list of all the containers and their stats.
- Return type:
List[Dict[str, str]]
- pyninja.features.dockerized.get_all_images() Optional[Dict[str, str]] ¶
Get all docker images.
- Returns:
Returns a dictionary with image stats.
- Return type:
Dict[str, str]
- pyninja.features.dockerized.get_all_volumes() Optional[Dict[str, str]] ¶
Get all docker volumes.
- Returns:
Returns a dictionary with list of volume objects.
- Return type:
Dict[str, str]
GPU¶
- pyninja.features.gpu._darwin(lib_path) Optional[List[Dict[str, str]]] ¶
Get GPU model and vendor information for Linux operating system.
- Parameters:
lib_path – Library path to get GPU information.
- Returns:
Returns a list of GPU model and vendor information.
- Return type:
List[Dict[str, str]]
- pyninja.features.gpu._linux(lib_path) Optional[List[Dict[str, str]]] ¶
Get GPU model and vendor information for Linux operating system.
- Parameters:
lib_path – Library path to get GPU information.
- Returns:
Returns a list of GPU model and vendor information.
- Return type:
List[Dict[str, str]]
- pyninja.features.gpu._windows(lib_path: Path) Optional[List[Dict[str, str]]] ¶
Get GPU model and vendor information for Windows operating system.
- Parameters:
lib_path – Library path to get GPU information.
- Returns:
Returns a list of GPU model and vendor information.
- Return type:
List[Dict[str, str]]
- pyninja.features.gpu.get_names() List[Dict[str, str]] ¶
Get list of GPU model and vendor information based on the operating system.
Operations¶
- pyninja.features.operations.default(name: str)¶
Default values for processes and services.
- pyninja.features.operations.get_process_info(proc: Process, process_name: str = None) Dict[str, str | int] ¶
Get process information.
- Parameters:
proc – Takes a
psutil.Process
object as an argument.process_name – Takes a custom process name as an optional argument.
- Returns:
Returns a dictionary with process usage statistics.
- Return type:
Dict[str, str | int]
- async pyninja.features.operations.process_monitor(processes: List[str]) List[Dict[str, str]] ¶
Function to monitor processes and return their usage statistics.
- Parameters:
processes – List of process names to monitor.
See also
Process names can be case in-sensitive as they are not strictly matched.
macOS/Linux: top | grep {{ process_name }}
Windows: Task Manager
- Returns:
Returns a list of dictionaries with process usage statistics.
- Return type:
List[Dict[str, str]]
- async pyninja.features.operations.service_monitor(services: List[str]) List[Dict[str, str]] ¶
Function to monitor services and return their usage statistics.
- Parameters:
services – List of service names to monitor.
See also
Service names are case-sensitive as they are strictly matched. Use the following command to get the right name.
macOS: launchctl list | grep {{ service_name }}
Linux: systemctl show {{ service_name }} –property=MainPID
Windows: sc query {{ service_name }}
- Returns:
Returns a list of dictionaries with service usage statistics.
- Return type:
List[Dict[str, str]]
- pyninja.features.operations.get_service_pid(service_name: str) Optional[int] ¶
Retrieve the PID of a service depending on the OS.
- pyninja.features.operations.get_service_pid_linux(service_name: str) Optional[int] ¶
Get the PID of a service on Linux.
- Parameters:
service_name – Name of the service.
- Returns:
Returns the PID of the service.
- Return type:
Optional[int]
- pyninja.features.operations.get_service_pid_macos(service_name: str) Optional[int] ¶
Get the PID of a service on macOS.
- Parameters:
service_name – Name of the service.
- Returns:
Returns the PID of the service.
- Return type:
Optional[int]
- pyninja.features.operations.get_service_pid_windows(service_name: str) Optional[int] ¶
Get the PID of a service on Windows.
- Parameters:
service_name – Name of the service.
- Returns:
Returns the PID of the service.
- Return type:
Optional[int]
Process¶
- pyninja.features.process.get_process_status(process_name: str, cpu_interval: int) List[Dict[str, int | float | str | bool]] ¶
Get process information by name.
- Parameters:
process_name – Name of the process.
cpu_interval – CPU interval to get the CPU performance.
- Returns:
Returns a list of performance report for each process hosting the given process name.
- Return type:
List[Dict[str, int | float | str | bool]]
- pyninja.features.process.get_performance(process: Process, cpu_interval: int) Dict[str, int | float | str | bool] ¶
Checks process performance by monitoring CPU utilization, number of threads and open files.
- Parameters:
process – Process object.
cpu_interval – CPU interval to get the CPU performance.
- Returns:
Returns the process metrics as key-value pairs.
- Return type:
Dict[str, int | float | str | bool]
Service¶
- pyninja.features.service.running(service_name: str) ServiceStatus ¶
Constructs an ServiceStatus object with a status code.
- Parameters:
service_name – Name of the service.
- Returns:
Returns a reference to the ServiceStatus object.
- Return type:
- 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:
- 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:
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:
- 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:
PyNinja - Modules¶
Exceptions¶
- exception pyninja.modules.exceptions.APIResponse(status_code: int, detail: Any = None, headers: Optional[Dict[str, str]] = None)¶
Custom
HTTPException
fromFastAPI
to wrap an API response.>>> APIResponse
- exception pyninja.modules.exceptions.UnSupportedOS¶
Custom exception class for unsupported OS.
>>> UnSupportedOS
- exception pyninja.modules.exceptions.RedirectException(location: str, detail: Optional[str] = '')¶
Custom
RedirectException
raised within the API since HTTPException doesn’t support returning HTML content.>>> RedirectException
See also
RedirectException allows the API to redirect on demand in cases where returning is not a solution.
There are alternatives to raise HTML content as an exception but none work with our use-case with JavaScript.
This way of exception handling comes handy for many unexpected scenarios.
References
https://fastapi.tiangolo.com/tutorial/handling-errors/#install-custom-exception-handlers
- exception pyninja.modules.exceptions.SessionError(detail: Optional[str] = '')¶
Custom exception class for session errors.
>>> SessionError
- pyninja.modules.exceptions.raise_os_error(operating_system: str) NoReturn ¶
Raises a custom exception for unsupported OS.
- Parameters:
operating_system – Current operating system.
- Raises:
ValidationError – Overridden exception from
pydantic.ValidationError
for unsupported OS.
Models¶
- class pyninja.modules.models.ServiceStatus(BaseModel)¶
Object to load service status with a status code and description.
>>> ServiceStatus
- status_code: int¶
- description: str¶
- class pyninja.modules.models.DiskLib(BaseModel)¶
Default disks library dedicated to each supported operating system.
>>> DiskLib
- linux: Path¶
- darwin: Path¶
- windows: Path¶
- class pyninja.modules.models.ServiceLib(BaseModel)¶
Default service library dedicated to each supported operating system.
>>> ServiceLib
- linux: Path¶
- darwin: Path¶
- windows: Path¶
- class pyninja.modules.models.ProcessorLib(BaseModel)¶
Default processor library dedicated to each supported operating system.
>>> ProcessorLib
- linux: Path¶
- darwin: Path¶
- windows: Path¶
- class pyninja.modules.models.Session(BaseModel)¶
Object to store session information.
>>> Session
- auth_counter: Dict[str, int]¶
- forbid: Set[str]¶
- info: Dict[str, str]¶
- rps: Dict[str, int]¶
- allowed_origins: Set[str]¶
- class pyninja.modules.models.RateLimit(BaseModel)¶
Object to store the rate limit settings.
>>> RateLimit
- max_requests: int¶
- seconds: int¶
- class pyninja.modules.models.EnvConfig(BaseModel)¶
Object to load environment variables.
>>> EnvConfig
- apikey: str¶
- ninja_host: str¶
- ninja_port: int¶
- remote_execution: bool¶
- api_secret: str | None¶
- monitor_username: str | None¶
- monitor_password: str | None¶
- monitor_session: int¶
- max_connections: int¶
- no_auth: bool¶
- processes: List[str]¶
- services: List[str]¶
- gpu_lib: Path¶
- disk_lib: Path¶
- service_lib: Path¶
- processor_lib: Path¶
- database: str¶
- log_config: Optional[Union[Dict[str, Any], Path]]¶
- classmethod parse_api_secret(value: str | None) str | None ¶
Parse API secret to validate complexity.
- Parameters:
value – Takes the user input as an argument.
- Returns:
Returns the parsed value.
- Return type:
str
- pyninja.modules.models.complexity_checker(secret: str) None ¶
Verifies the strength of a secret.
See also
A secret is considered strong if it at least has:
32 characters
1 digit
1 symbol
1 uppercase letter
1 lowercase letter
- Raises:
AssertionError – When at least 1 of the above conditions fail to match.
- class pyninja.modules.models.GPULib(*, linux: Path = '/usr/bin/lspci', darwin: Path = '/usr/sbin/system_profiler', windows: Path = 'C:\\Windows\\System32\\wbem\\wmic.exe')¶
Default GPU library dedicated to each supported operating system.
>>> GPULib
- linux: Path¶
- darwin: Path¶
- windows: Path¶
- _abc_impl = <_abc._abc_data object>¶
- model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_fields: ClassVar[dict[str, FieldInfo]] = {'darwin': FieldInfo(annotation=Path, required=False, default='/usr/sbin/system_profiler', metadata=[PathType(path_type='file')]), 'linux': FieldInfo(annotation=Path, required=False, default='/usr/bin/lspci', metadata=[PathType(path_type='file')]), 'windows': FieldInfo(annotation=Path, required=False, default='C:\\Windows\\System32\\wbem\\wmic.exe', metadata=[PathType(path_type='file')])}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].
This replaces Model.__fields__ from Pydantic V1.
- class pyninja.modules.models.WSSession(*, invalid: Dict[str, int] = {}, client_auth: Dict[str, Dict[str, int]] = {})¶
Object to store websocket session information.
>>> WSSession
- invalid: Dict[str, int]¶
- client_auth: Dict[str, Dict[str, int]]¶
- _abc_impl = <_abc._abc_data object>¶
- model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}¶
A dictionary of computed field names and their corresponding ComputedFieldInfo objects.
- model_config: ClassVar[ConfigDict] = {}¶
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
- model_fields: ClassVar[dict[str, FieldInfo]] = {'client_auth': FieldInfo(annotation=Dict[str, Dict[str, int]], required=False, default={}), 'invalid': FieldInfo(annotation=Dict[str, int], required=False, default={})}¶
Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].
This replaces Model.__fields__ from Pydantic V1.
- pyninja.modules.models.get_library(library: Union[Type[ServiceLib], Type[ProcessorLib], Type[DiskLib], Type[GPULib]]) Path ¶
Get service/processor/disk library filepath for the host operating system.
- Parameters:
library – Library class inherited from BaseModel.
- Returns:
Returns the
FilePath
referencing the appropriate library.- Return type:
FilePath
- class pyninja.modules.models.Database(datastore: Union[Path, str], timeout: int = 10)¶
Creates a connection to the Database using sqlite3.
>>> Database
- create_table(table_name: str, columns: Union[List[str], Tuple[str]]) None ¶
Creates the table with the required columns.
- Parameters:
table_name – Name of the table that has to be created.
columns – List of columns that has to be created.
Payloads¶
- class pyninja.modules.payloads.RunCommand(BaseModel)¶
Payload for run-command endpoint.
>>> RunCommand
- command: str¶
- timeout: Union[int, float]¶
- class pyninja.modules.payloads.ListFiles(BaseModel)¶
Payload for list-files endpoint.
>>> ListFiles
- directory: Path¶
- include_directories: bool¶
- deep_scan: bool¶
RateLimit¶
- class pyninja.modules.rate_limit.RateLimiter(rps: RateLimit)¶
Object that implements the
RateLimiter
functionality.>>> RateLimiter
- init(request: Request) None ¶
Checks if the number of calls exceeds the rate limit for the given identifier.
- Parameters:
request – The incoming request object.
- Raises:
429 – Too many requests.
Secure¶
- async pyninja.modules.secure.calculate_hash(value: Any) str ¶
Generate hash value for the given payload.
- async pyninja.modules.secure.base64_encode(value: Any) str ¶
Base64 encode the given payload.
- async pyninja.modules.secure.base64_decode(value: Any) str ¶
Base64 decode the given payload.
- async pyninja.modules.secure.hex_decode(value: Any) str ¶
Convert hex value to a string.
- async pyninja.modules.secure.hex_encode(value: str)¶
Convert string value to hex.
PyNinja - Monitor¶
Authenticator¶
- async pyninja.monitor.authenticator.failed_auth_counter(host) None ¶
Keeps track of failed login attempts from each host, and redirects if failed for 3 or more times.
- Parameters:
host – Host header from the request.
- async pyninja.monitor.authenticator.raise_error(host: str) NoReturn ¶
Raises a 401 Unauthorized error in case of bad credentials.
- Parameters:
host – Host header from the request.
- async pyninja.monitor.authenticator.extract_credentials(authorization: HTTPAuthorizationCredentials, host: str) List[str] ¶
Extract the credentials from
Authorization
headers and decode it before returning as a list of strings.- Parameters:
authorization – Authorization header from the request.
host – Host header from the request.
- async pyninja.monitor.authenticator.verify_login(authorization: HTTPAuthorizationCredentials, host: str) Dict[str, str | int] ¶
Verifies authentication and generates session token for each user.
- Returns:
Returns a dictionary with the payload required to create the session token.
- Return type:
Dict[str, str]
- async pyninja.monitor.authenticator.generate_cookie(auth_payload: dict) Dict[str, str | bool | int] ¶
Generate a cookie for monitoring page.
- Parameters:
auth_payload – Authentication payload containing username and timestamp.
- Returns:
Returns a dictionary with cookie details
- Return type:
Dict[str, str | bool | int]
- async pyninja.monitor.authenticator.session_error(request: Request, error: SessionError) HTMLResponse ¶
Renders the session error page.
- Parameters:
request – Reference to the FastAPI request object.
error – Session error message.
- Returns:
Returns an HTML response templated using Jinja2.
- Return type:
HTMLResponse
- async pyninja.monitor.authenticator.validate_session(host: str, cookie_string: str, log: bool = True) None ¶
Validate the session token.
- Parameters:
host – Hostname from the request.
cookie_string – Session token from the cookie.
log – Boolean flag to enable logging.
- Raises:
Raises a SessionError with summary. –
Configuration¶
- async pyninja.monitor.config.clear_session(response: HTMLResponse) HTMLResponse ¶
Clear the session token from the response.
- Parameters:
response – Takes the
Response
object as an argument.- Returns:
Returns the response object with the session token cleared.
- Return type:
Response
- async pyninja.monitor.config.get_expiry(lease_start: int, lease_duration: int) str ¶
Get expiry datetime as string using max age.
- Parameters:
lease_start – Time when the authentication was made.
lease_duration – Number of seconds until expiry.
- Returns:
Returns the date and time of expiry in GMT.
- Return type:
str
Resources¶
- pyninja.monitor.resources.landing_page() Dict[str, Any] ¶
Returns the landing page context for monitor endpoint.
- Returns:
Returns a key-value pair to be inserted into the Jinja template.
- Return type:
Dict[str, Any]
- pyninja.monitor.resources.container_cpu_limit(container_name: str) int | None ¶
Get CPU cores configured for a particular container using NanoCpus.
- Parameters:
container_name – Name of the docker container.
- Returns:
Returns the number of CPU cores.
- Return type:
int
- pyninja.monitor.resources.map_docker_stats(json_data: Dict[str, str]) Dict[str, str] ¶
Map the JSON data to a dictionary.
- Parameters:
json_data – JSON data from the docker stats command.
- Returns:
Returns a dictionary with container stats.
- Return type:
Dict[str, str]
- pyninja.monitor.resources.get_cpu_percent(cpu_interval: int) List[float] ¶
Get CPU usage percentage.
- Parameters:
cpu_interval – Interval to get the CPU performance.
- Returns:
Returns a list of CPU percentages.
- Return type:
List[float]
- pyninja.monitor.resources.containers() bool ¶
Check if any Docker containers are running.
- async pyninja.monitor.resources.get_docker_stats() List[Dict[str, str]] ¶
Run the docker stats command asynchronously and parse the output.
- Returns:
Returns a list of key-value pairs with the container stat and value.
- Return type:
List[Dict[str, str]]
- async pyninja.monitor.resources.get_system_metrics() Dict[str, dict] ¶
Async handler for virtual memory, swap memory disk usage and CPU load averages.
- Returns:
Returns a nested dictionary.
- Return type:
Dict[str, dict]
- async pyninja.monitor.resources.system_resources() Dict[str, dict] ¶
Gather system resources including Docker stats asynchronously.
- Returns:
Returns a nested dictionary.
- Return type:
Dict[str, dict]
Routes¶
- async pyninja.monitor.routes.error_endpoint(request: Request) HTMLResponse ¶
Error endpoint for the monitoring page.
- Parameters:
request – Reference to the FastAPI request object.
- Returns:
Returns an HTML response templated using Jinja2.
- Return type:
HTMLResponse
- async pyninja.monitor.routes.logout_endpoint(request: Request) HTMLResponse ¶
Logs out the user and clears the session token.
- Parameters:
request – Reference to the FastAPI request object.
- Returns:
Redirects to login page.
- Return type:
HTMLResponse
- async pyninja.monitor.routes.login_endpoint(request: Request, authorization: HTTPAuthorizationCredentials = Depends(HTTPBearer)) JSONResponse ¶
Login endpoint for the monitoring page.
- Returns:
Returns a JSONResponse object with a
session_token
andredirect_url
set.- Return type:
JSONResponse
- async pyninja.monitor.routes.monitor_endpoint(request: Request, session_token: str = Cookie(None))¶
Renders the UI for monitoring page.
- Parameters:
request – Reference to the FastAPI request object.
session_token – Session token set after verifying username and password.
- Returns:
Returns an HTML response templated using Jinja2.
- Return type:
HTMLResponse
- async pyninja.monitor.routes.websocket_endpoint(websocket: WebSocket, session_token: str = Cookie(None))¶
Websocket endpoint to fetch live system resource usage.
- Parameters:
websocket – Reference to the websocket object.
session_token – Session token set after verifying username and password.