pawnlib.resource package

from pawnlib.resource import *

Submodules

pawnlib.resource.net module

from pawnlib.resource import net

class ProcNetMonitor(top_n=10, refresh_rate=2, group_by='pid', unit='Mbps', protocols=None, pid_filter=None, proc_filter=None, min_bytes_threshold=0, callback=None, exit_signal='EXIT')

Bases: object

ProcNetMonitor monitors network usage of processes using eBPF.

Parameters:

• top_n (optional) – The top N processes to display in the table. |default| 10

• refresh_rate (optional) – The data refresh rate (refreshes per second). |default| 2

• group_by (optional) – How to group processes (“pid” or “name”). |default| 'pid'

• unit (optional) – The unit for displaying rates (e.g., “Mbps”). |default| 'Mbps'

• protocols (optional) – List of protocols to monitor. Defaults to [“tcp”, “udp”]. |default| None

• pid_filter (optional) – List of PIDs to monitor. Only these PIDs will be tracked. |default| None

• proc_filter (optional) – List of process names to monitor. Only these processes will be tracked. |default| None

• min_bytes_threshold (optional) – Minimum number of bytes to consider for monitoring. |default| 0

• callback (optional) – A user-defined function to be called when data is updated. |default| None

• exit_signal (optional) – A string used as the termination signal. When this string is detected, the monitoring process will stop and the program will exit. Default is “EXIT”. |default| 'EXIT'

Example:

# Example 1: Monitor top 5 processes with TCP and UDP protocols
def handle_update(data):
    for group, info in data.items():
        print(f"Group: {group}, Sent: {info['bytes_sent']} bytes, Recv: {info['bytes_recv']} bytes")

monitor = ProcNetMonitor(top_n=5, protocols=["tcp", "udp"], callback=handle_update)
monitor.run()

# Example 2: Monitor specific PIDs
def handle_specific_pids(data):
    for pid, info in data.items():
        print(f"PID: {pid}, Sent: {info['bytes_sent']} bytes, Recv: {info['bytes_recv']} bytes")

monitor = ProcNetMonitor(pid_filter=[1234, 5678], callback=handle_specific_pids)
monitor.run()

# Example 3: Monitor specific process names
def handle_specific_procs(data):
    for proc, info in data.items():
        print(f"Process: {proc}, Sent: {info['bytes_sent']} bytes, Recv: {info['bytes_recv']} bytes")

monitor = ProcNetMonitor(proc_filter=["python", "nginx"], callback=handle_specific_procs)
monitor.run()

# Example 4: Set a custom refresh rate and exit signal
def handle_exit(data):
    print("Received data, checking exit condition...")
    if some_condition:
        return 'EXIT'

monitor = ProcNetMonitor(refresh_rate=1, exit_signal="EXIT", callback=handle_exit)
monitor.run()

Initialize the ProcNetMonitor class.

Parameters:

• top_n (optional) – The top N processes to display in the table. |default| 10

• refresh_rate (optional) – The data refresh rate (refreshes per second). |default| 2

• group_by (optional) – How to group processes (“pid” or “name”). |default| 'pid'

• unit (optional) – The unit for displaying rates (e.g., “Mbps”). |default| 'Mbps'

• protocols (optional) – List of protocols to monitor. Defaults to [“tcp”, “udp”]. |default| None

• pid_filter (optional) – List of PIDs to monitor. Only these PIDs will be tracked. |default| None

• proc_filter (optional) – List of process names to monitor. Only these processes will be tracked. |default| None

• min_bytes_threshold (optional) – Minimum number of bytes to consider for monitoring. |default| 0

• callback (optional) – A user-defined function to be called when data is updated. |default| None

• exit_signal (optional) – A string used as the termination signal. When this string is detected, the monitoring process will stop and the program will exit. Default is “EXIT”. |default| 'EXIT'

EVENT_TCP_SEND = 1

EVENT_TCP_RECV = 2

EVENT_UDP_SEND = 3

EVENT_UDP_RECV = 4

__init__(top_n=10, refresh_rate=2, group_by='pid', unit='Mbps', protocols=None, pid_filter=None, proc_filter=None, min_bytes_threshold=0, callback=None, exit_signal='EXIT')

Initialize the ProcNetMonitor class.

Parameters:

• top_n (optional) – The top N processes to display in the table. |default| 10

• refresh_rate (optional) – The data refresh rate (refreshes per second). |default| 2

• group_by (optional) – How to group processes (“pid” or “name”). |default| 'pid'

• unit (optional) – The unit for displaying rates (e.g., “Mbps”). |default| 'Mbps'

• protocols (optional) – List of protocols to monitor. Defaults to [“tcp”, “udp”]. |default| None

• pid_filter (optional) – List of PIDs to monitor. Only these PIDs will be tracked. |default| None

• proc_filter (optional) – List of process names to monitor. Only these processes will be tracked. |default| None

• min_bytes_threshold (optional) – Minimum number of bytes to consider for monitoring. |default| 0

• callback (optional) – A user-defined function to be called when data is updated. |default| None

• exit_signal (optional) – A string used as the termination signal. When this string is detected, the monitoring process will stop and the program will exit. Default is “EXIT”. |default| 'EXIT'

setup_signal_handlers()

Set up signal handlers to perform cleanup on script termination.

static get_process_cmdline(pid)

Get the command-line arguments of a process by PID.

generate_title()

Dynamically generate the title based on initialized parameters.

Returns:

Generated title.

Return type:

str

run()

ProcNetMonitor를 실행합니다. `is_running`이 `True`인 동안 루프를 계속합니다.

run_live()

Display real-time tables using Rich Live.

update_data()

Poll eBPF events and update the process_network data.

get_latest_network_data(top_n=None)

Return the latest network usage data, optionally limited to the top N processes.

Parameters:

top_n (int) – Number of top processes to return. |default| None

Returns:

List of process network usage dictionaries.

Return type:

List[Dict[str, Any]]

get_top_n(n=None)

Retrieve the top N processes by network usage.

Parameters:

n (Optional[int], optional) – Number of top processes to retrieve. Defaults to self.top_n. |default| None

Return type:

List[Dict[str, Union[int, str, float]]]

Returns:

List of top N processes sorted by (bytes_sent + bytes_recv).

class OverrideDNS(domain='', ipaddr='', port=80)

Bases: object

Change the Domain Name using socket

Example

from pawnlib.resource import net
net.OverrideDNS(domain=domain, ipaddr=ipaddr).set()

new_getaddrinfo(*args)

set()

unset()

get_public_ip(use_cache=False, *, services=None, timeout=2.0, connect_timeout=None, raise_on_error=False)

Returns the public IP address of the current machine.

Parameters:

• use_cache (bool, optional) – Whether to use and update the cached public IP |default| False

• services (Optional[List[str]], optional) – List of public IP services to query in order |default| None

• timeout (float, optional) – Response timeout (seconds) |default| 2.0

• connect_timeout (Optional[float], optional) – Connection timeout (seconds). If None, use timeout. |default| None

• raise_on_error (bool, optional) – If True, re-raise the last exception instead of returning an empty string |default| False

Return type:

str

Returns:

The public IP address, or empty string on failure (if raise_on_error=False)

Example:

from pawnlib.resource import net

net.get_public_ip()
net.get_public_ip(use_cache=True)

class FindFastestRegion(verbose=True, aws_regions=None)

Bases: object

run()

async find_fastest_region()

get_time(url, name='NULL')

sorted_results(key='run_time')

print_results()

class AsyncPortScanner(ip_range, port_range=(0, 65535), max_concurrency=30, timeout=1, ping_timeout=0.05, fast_scan_ports=[22, 80, 443], batch_size=50000)

Bases: object

Asynchronous Port Scanner class.

Parameters:

• ip_range (Tuple[str, str]) – Tuple of start and end IP addresses to scan.

• port_range (Tuple[int, int], optional) – Tuple of start and end ports to scan. Default is all ports (0, 65535). |default| (0, 65535)

• max_concurrency (int, optional) – Maximum number of concurrent scans. Default is 30. |default| 30

Example

scanner = AsyncPortScanner(("192.168.0.1", "192.168.0.255"), (1, 1024), 50)
asyncio.run(scanner.scan_all())

async ping_host(ip)

Return type:

bool

async try_ping_host(ip, progress, task_id)

async scan_all(fast_scan=False)

async check_and_scan_host(ip)

async scan_port(ip, port)

Return type:

(str, int, bool)

calculate_scan_range()

async scan(fast_scan=False, progress=None)

async get_ips_to_scan(fast_scan, progress)

Return type:

List[str]

async wrap_scan(ip, port, progress, task_id)

static ip_to_int(ip)

Return type:

int

static int_to_ip(ip_int)

Return type:

str

get_results()

print_scan_results(view='all')

run_scan(fast_scan=False)

get_local_ip()

Get the local IP address

Returns:

Example

from pawnlib.resource import net
net.get_local_ip()

get_hostname()

Get the local hostname

Returns:

Example

from pawnlib.resource import net
net.get_hostname()

extract_host_port(host)

The extract_host_port function extracts the host and port from a string.

Parameters:

host – Extract the hostname from the url

Returns:

A tuple of the host and port number

Example:

from pawnlib.resource import net
net.extract_host_port("http://127.0.0.1:8000")

check_port(host='', port=0, timeout=3.0, protocol='tcp')

Returns boolean with checks if the port is open

Parameters:

• host (str, optional) – ipaddress os hostname |default| ''

• port (int, optional) – destination port number |default| 0

• timeout (float, optional) – timeout sec |default| 3.0

• protocol (Literal['tcp', 'udp'], optional) – type of protocol |default| 'tcp'

Return type:

bool

Returns:

boolean

Example

from pawnlib.resource import net
net.check_port()

listen_socket(host, port)

Create a socket object and bind it to the host and port provided. Listen for incoming connections on that socket, with a maximum of 5 connections in the queue.

Parameters:

• host – str - hostname of the machine where the server is running

• port – int - port number that the server will listen on

Returns:

socket - a socket object

Example

# create a socket object and bind it to localhost and port 8080
sock = listen_socket("localhost", 8080)

# listen for incoming connections
conn, addr = sock.accept()

wait_for_port_open(host='', port=0, timeout=3.0, protocol='tcp', sleep=1)

Wait for a port to open. Useful when writing scripts which need to wait for a server to be available.

Parameters:

• host (str, optional) – hostname or ipaddress |default| ''

• port (int, optional) – port |default| 0

• timeout (float, optional) – timeout seconds (float) |default| 3.0

• protocol (Literal['tcp', 'udp'], optional) – tcp or udp |default| 'tcp'

• sleep (float, optional) – sleep fime seconds (float) |default| 1

Return type:

bool

Returns:

Example

from pawnlib.resource import net
net.wait_for_port_open("127.0.0.1", port)

## ⠏  Wait for port open 127.0.0.1:9900 ... 6

get_location(ipaddress='')

get_location_with_ip_api()

pawnlib.resource.server module

from pawnlib.resource import server

hex_mask_to_cidr(hex_mask)

Converts a hexadecimal mask to a CIDR value.

Parameters:

hex_mask – The hexadecimal mask to convert.

Returns:

The CIDR value of the mask.

Example

from pawnlib.resource import server

server.hex_mask_to_cidr("FFFF")
# >> 16

server.hex_mask_to_cidr("FF00")
# >> 8

get_interface_names()

Get a list of interface names on the system.

Returns:

a list of interface names

Example

from pawnlib.resource import server
server.get_interface_names()
# >> ['lo', 'enp0s31f6', 'wlp0s20f3']

get_ip_addresses(interface)

Get a list of IP addresses associated with a given network interface.

Parameters:

interface – A string representing the name of the network interface.

Returns:

A list of IP addresses associated with the given network interface.

Example

from pawnlib.resource import server
server.ip_addresses = get_ip_addresses('eth0')
# >> ['192.168.0.1', '192.168.0.2']

get_ip_and_netmask(interface)

Get IP address and netmask of the given interface.

Parameters:

interface – The name of the interface.

Returns:

A list containing the IP address and netmask.

Example

from pawnlib.resource import server
server.ip_info = get_ip_and_netmask('eth0')
print(ip_info)
# >> ['192.168.0.2', '255.255.255.0']

subnet_mask_to_decimal(subnet_mask)

Convert subnet mask to decimal.

Parameters:

subnet_mask (int) – Subnet mask in integer format.

Returns:

Decimal subnet mask.

Return type:

str

Example

from pawnlib.resource import server
server.subnet_mask_to_decimal(24)
# >> '255.255.255.0'

server.subnet_mask_to_decimal(16)
# >> '255.255.0.0'

get_interface_ips(ignore_interfaces=None, detail=False, is_sort=True, ip_only=False)

Get the IP addresses of the interfaces.

Parameters:

• ignore_interfaces (optional) – A list of interface names to ignore. |default| None

• detail (optional) – Whether to show detailed information or not. |default| False

• is_sort (optional) – Whether to sort the results or not. |default| True

• ip_only (optional) – If True, return only the IP addresses. |default| False

Returns:

A list of tuples containing interface name and IP address, a dictionary with IP, subnet, and gateway, or just IPs if ip_only is True.

Example

from pawnlib.resource import server

server.get_interface_ips()
# >> [('lo', '127.0.0.1 / 8'), ('wlan0', '192.168.0.10, G/W: 192.168.0.1')]

server.get_interface_ips(detail=True)
# >>  [ ('en0', {'ip': '20.22.1.13', 'subnet': '255.255.252.0', 'gateway': '20.22.0.1'}),('utun4', {'ip': '43.62.13.6'})]

server.get_interface_ips(ip_only=True)
# >> ['127.0.0.1', '192.168.0.10']

get_interface_ips_dict(ignore_interfaces=[])

Get the IP addresses of all interfaces in a dictionary format.

Parameters:

ignore_interfaces (optional) – list of interfaces to be ignored |default| []

Returns:

dictionary with interface names as keys and their IP addresses as values

Example

from pawnlib.resource import server

server.get_interface_ips_dict(ignore_interfaces=['lo', 'eth0'])
# >> {'wlan0': '192.168.1.100'}

get_default_route_and_interface()

Parse the route of the process based on the platform.

Example

from pawnlib.resource import server

get_default_route_and_interface()
# If Linux:
# >> ('192.168.1.1', 'eth0')
# If MacOS:
# >> ('192.168.1.1', 'en0')
# If other platform:
# >> ("Unsupported platform.", None)

get_default_route_and_interface_linux()

Parse the Linux route.

Example

from pawnlib.resource import server

server.get_default_route_and_interface_linux()
# >> ('192.168.1.1', 'eth0')

get_default_route_and_interface_macos()

Parse the Linux route.

Example

from pawnlib.resource import server

server.get_default_route_and_interface_macos()
# >> ('192.168.1.1', 'eth0')

class ProcessMonitor(console=None, n=5)

Bases: object

stop()

모든 모니터링 작업을 종료합니다.

get_top_processes(n=5, resource='memory')

Return type:

List[Dict[str, Union[str, float]]]

get_network_usage_from_monitor()

Fetch network usage data from ProcNetMonitor.

Returns:

Process network usage data.

Return type:

List[Dict[str, Union[int, str, float]]]

create_resource_table(processes, resource)

리소스 데이터에 대한 리치 테이블 생성.

Return type:

Table

create_dashboard()

Return type:

Layout

run_live()

실시간 대시보드를 표시합니다.

class MemoryStatus(proc_path='/proc')

Bases: object

static read_stats_file(file_path)

Return type:

str

static parse_meminfo(meminfo)

Return type:

Dict[str, int]

get_memory_pressure()

Return type:

Dict[str, float]

static get_top_memory_processes(n=5)

Return type:

List[Dict[str, Union[str, float]]]

get_huge_pages_info()

Return type:

Dict[str, int]

get_memory_status(unit='GB', output_format='dict')

Return type:

Union[Dict, str]

check_memory_threshold(threshold=90.0)

메모리 사용량이 임계값을 초과하는지 확인

Return type:

bool

class SystemMonitor(interval=1, proc_path='/proc')

Bases: object

static read_stats_file(filename='')

parse_net_dev()

parse_cpu_stat()

get_cpu_status(decimal=1)

collect_system_status()

get_network_status()

read_disk_stats()

get_disk_usage()

get_memory_status(unit='GB')

get_system_status()

print_memory_status()

get_netstat_count(proc_path='/proc', detail=False)

convert_hex_to_ip_port(address)

get_rlimit_nofile(detail=False)

Returns a dict with the current soft and hard limits of resource. If detail is True, it also includes the number of used file handles and their usage percentage.

Parameters:

detail (optional) – If True, include used_handles and usage_percentage. |default| False

Returns:

A dictionary containing ‘soft’, ‘hard’, and optionally ‘used_handles’, ‘usage_percentage’.

Example

from pawnlib.resource import server
server.get_rlimit_nofile(detail=True)

## > {'soft': 1024, 'hard': 4096, 'used_handles': 512, 'usage_percentage': 50.0}

get_mac_platform_info()

Returns:

Example

from pawnlib.resource import server
server.get_mac_platform_info()

get_platform_info(**kwargs)

Returns a dict with platform information, including the specific operating system.

Returns:

A dictionary containing platform information.

Example

from pawnlib.resource import server
server.get_platform_info()

## > {'system': 'Darwin', 'os': 'macOS', 'version': 'Darwin Kernel Version 21.6.0: Wed Aug 10 14:28:23 PDT 2022; root:xnu-8020.141.5~2/RELEASE_ARM64_T6000', 'release': '21.6.0', 'machine': 'arm64', 'processor': 'arm', 'python_version': '3.9.13', 'model': 'Apple M1 Pro', 'cores': 10}

parse_cpu_load(load_str)

get_cpu_load()

Returns dict with current cpu load average :return:

Example

from pawnlib.resource import server
server.get_cpu_load()
## > {'1min': 12.29, '5min': 11.01, '15min': 11.09}

parse_proc_stat()

calculate_iowait_linux(interval=1)

run_command(command)

get_iowait()

get_uptime()

get_swap_usage()

get_macos_swap_usage()

get_linux_swap_usage()

parse_meminfo_line(meminfo, key)

format_swap_usage(used_swap_gb, total_swap_gb)

get_uptime_cmd()

Returns dict with current cpu load average using uptime command :rtype: dict :return:

Example

from pawnlib.resource import server
server.get_uptime_cmd()
## > {'1min': 12.29, '5min': 11.01, '15min': 11.09}

get_load_average()

get_total_memory_usage()

Returns float with current memory usage using ps command :rtype: float :return: kilo bytes

Example

from pawnlib.resource import server
server.get_total_memory_usage()

## > 24246272.0

get_mem_osx_info()

get_mem_info(unit='GB')

Read in the /proc/meminfo and return a dictionary of the memory and swap usage for all processes.

convert_values_to_unit(data, convert_unit)

get_cpu_time()

Reads /proc/stat to gather CPU time statistics for each CPU core. Returns a dictionary with CPU IDs and their respective time metrics.

Return type:

Dict[str, Dict[str, float]]

get_cpu_usage_percentage(interval=1.0)

Calculates CPU usage percentage over a given interval. Returns a dictionary with CPU usage percentages per core, average, and iowait.

Return type:

Dict[str, float]

get_aws_metadata(meta_ip='169.254.169.254', timeout=2)

aws_data_crawl(url, d, timeout)

get_kakao_metadata(meta_ip='169.254.169.254', timeout=2.0)

Retrieves metadata from Kakao Cloud instance.

Parameters:

• meta_ip (str) – IP address of the metadata service |default| '169.254.169.254'

• timeout (float) – Timeout for HTTP requests in seconds |default| 2.0

Returns:

Metadata from Kakao Cloud instance

Return type:

dict

Raises:

MetadataError – When metadata retrieval fails

get_gcp_metadata(meta_ip='metadata.google.internal', timeout=2)

async get_oci_metadata_async(meta_ip='169.254.169.254', timeout=2)

get_oci_metadata(meta_ip='169.254.169.254', timeout=2)

io_flags_to_string(flags)

class DiskUsage

Bases: object

match_list(patterns, text)

Check if the text matches any pattern in the list.

get_mount_points()

Get a list of all mount points along with their device names.

get_disk_usage(mount_point='/', unit='GB', precision=2)

Get disk usage information for a specific mount point or all mount points.

Parameters:

• mount_point (optional) – Mount point to check. Use “/”, “/home”, or “all”. |default| '/'

• unit (optional) – Unit for disk usage. Can be “B”, “KB”, “MB”, “GB”, “TB”, or “auto”. Default is “GB”. |default| 'GB'

• precision (optional) – Number of decimal places for the output. Default is 2. |default| 2

Returns:

Disk usage information.

static calculate_disk_usage(mount_point, factor, precision, unit)

Helper function to calculate disk usage for a given mount point.

calculate_disk_usage_with_auto_unit(mount_point, precision, unit)

Calculate disk usage with automatic unit selection if ‘auto’ is specified.

class DiskPerformanceTester(file_path, file_size_mb, iterations=5, block_size_kb=1024, num_threads=1, io_pattern='sequential', decimal_places=2, console=None, verbose=False, additional_info=None)

Bases: object

Class to test disk performance by measuring read and write speeds.

Parameters:

• file_path – Path to the file used for testing.

• file_size_mb – Size of the test file in megabytes.

• iterations (optional) – Number of iterations for the test. |default| 5

• block_size_kb (optional) – Size of each block in kilobytes. |default| 1024

• num_threads (optional) – Number of threads to use for the test. |default| 1

• io_pattern (optional) – I/O pattern, e.g., “sequential” or “random”. |default| 'sequential'

• decimal_places (optional) – Number of decimal places for results. |default| 2

• console (optional) – Console object for logging. |default| None

• verbose (optional) – Flag to enable verbose logging. |default| False

• additional_info (optional) – Additional info for result file |default| None

Example

tester = DiskPerformanceTester("/tmp/testfile", 100)

tester.run_parallel_tests()

# or

tester.measure_write_speed("/tmp/testfile", task_id, progress)
tester.measure_read_speed("/tmp/testfile", task_id, progress)
tester.cleanup_and_exit()

log_with_progress(progress, task_id, message)

Helper function to log messages with progress

get_write_flags()

measure_write_speed(file_path, task_id, progress)

measure_read_speed(file_path, task_id, progress)

prepare_file(file_path, task_id, progress)

cleanup(file_path)

cleanup_files()

cleanup_and_exit(signum, frame)

run_test(task_name, measure_func, progress)

run_parallel_tests()

validate_speeds(speeds)

print_summary()

create_results_dict()

save_results()

class FileSystemTester(test_dir, file_count=1000, file_size=1024, iterations=5, decimal_places=3, verbose=False, console=None)

Bases: object

setup()

cleanup()

generate_random_string(size)

test_file_creation(progress, task_id)

test_file_reading(progress, task_id)

test_file_deletion(progress, task_id)

test_directory_traversal(progress, task_id)

run_tests(is_print=False)

print_results(results)

class SSHLogPathResolver(os_name=None, raise_on_failure=False)

Bases: object

log_file_mapping = {'amzn': '/var/log/secure', 'centos': '/var/log/secure', 'debian': '/var/log/auth.log', 'fedora': '/var/log/secure', 'macos': '/var/log/system.log', 'rocky': '/var/log/secure', 'ubuntu': '/var/log/auth.log'}

get_path()

extract_directory()