darknet-toolkit

An enterprise-grade Python utility toolkit — batteries-included helpers for system introspection, networking, files, security, text, dates, developer tooling, web APIs, data wrangling, cryptography, storage, concurrent compute and OS automation. One dependency-free core, one consistent error model, one hardened CLI.

$ dark-net system info
$ dark-net network dns example.com --json
$ dark-net file hash ./release.tar.gz --algorithm sha512
$ dark-net security audit --json

• Zero runtime dependencies. Everything in the standard library; optional extras (psutil, PyYAML, pip-audit) unlock richer platform data.

• Safe by default. Deny-by-default SSRF guards, TLS verification that cannot be weakened implicitly, consent-gated destructive operations, secret redaction in every human-facing output path.

• One error model. Every failure raises a subclass of DarkNetError with a code, a human message and machine-readable details.

• Predictable exit codes. The CLI maps failure classes to stable exit codes (3 validation, 4 not found, 6 network, 9 consent required, 11 check failed, …), so scripts can branch on the outcome.

• Typed and tested. py.typed ships in the wheel; mypy --strict clean on src/; 2,100+ tests, 92 % line coverage, no network required for the default suite.

• Flat by design. All 620 public names are importable straight from the package root: dark_net.decrypt(...) is exactly dark_net.crypto.decrypt(...).

---

Contents

• Installation
• Quick start
• Command-line interface
• Library tour
  • Configuration
  • Errors and exit codes
  • System
  • Network
  • Files
  • Security
  • Text
  • Dates and times
  • Developer tooling
  • Web and HTTP
  • Data
  • Crypto
  • Storage
  • Concurrency
  • Automation
• Security posture
• Optional extras
• Development
• Documentation
• License

---

Installation

python -m pip install darknet-toolkit        # core, no third-party dependencies
python -m pip install "darknet-toolkit[all]" # core + psutil, PyYAML, httpx, pip-audit

Requires Python 3.10+. The package is pure Python and works on Linux, macOS and Windows; a few platform-specific fields (hardware inventory, physical core count) are only populated where the OS exposes them.

From a checkout:

python -m pip install -e ".[dev]"

Quick start

Library

Snippets assume this repository as the working directory (they hash pyproject.toml and write state.json next to it).

from dark_net import get_config
from dark_net.files import hash_file, read_json, write_json
from dark_net.network import parse_url, resolve_hostname
from dark_net.security import generate_password, redact

config = get_config()  # layered: defaults → file → env → overrides
print(config.timeout, config.verify_tls)

print(hash_file("pyproject.toml", algorithm="sha512")[:32])  # any supported algorithm
write_json("state.json", {"ok": True}, atomic=True)
print(read_json("state.json"))

print(resolve_hostname("example.com").addresses)  # never raises; reports errors
url = parse_url("https://example.com:8443/a?b=1")  # credentials are rejected
print(url.host, url.effective_port)  # example.com 8443

print(redact("token=abc123 password=hunter2"))  # token=*** password=***
print(generate_password(24))

# flat namespace: every public name is one import away
from dark_net import encrypt, parallel_map, exec_command, shannon_entropy

print(round(shannon_entropy(b"ab"), 1))  # 1.0

CLI

$ dark-net --version
dark-net 1.1.0

$ dark-net system cpu --json
$ dark-net network url "https://example.com:8443/path?x=1"
$ dark-net file verify ./artifact.bin 9f86d081...   # exit 11 on mismatch
$ echo 'password=hunter2' | dark-net security redact

Command-line interface

Global flags come before the subcommand (or after it — both work):

Flag	Meaning
-q, --quiet	Suppress human-readable output
--json	Emit machine-readable JSON on stdout
--log-level {DEBUG,…,CRITICAL}	Logging verbosity
--no-redact	Disable secret redaction (not recommended)
--version / -h, --help	Version / help

Command groups:

Group	Subcommands
system	info [--hardware], platform, cpu, memory, disk [PATH], processes [--limit N] [--sort pid|name|memory_rss|cpu_percent], env [--prefix P] [--show-values]
network	info, dns HOST [--type A|AAAA|ANY], ip ADDRESS, url URL, check [--port N], port HOST PORT [--yes], public-ip [--provider NAME]
file	hash PATH [--algorithm …], info PATH, duplicates PATH, verify PATH EXPECTED, search [--pattern …] [--max-results N], manifest PATH [--output FILE]
security	audit [--path P] [--no-dependencies], token, password, strength [PASSWORD|--stdin], redact [TEXT|--file F]
text	slug TEXT, stats [TEXT|--file F], case TEXT --to {snake,kebab,camel,…}
datetime	now [--tz TZ] [--format F], parse VALUE, diff START END, duration [SECONDS] [--style short|long|clock]
config	show, validate
doctor	[--strict] — dependency, configuration and reachability diagnostics
crypto	keygen [--length N] [--hex], digest TEXT|--file P [--algorithm …], encrypt TEXT --key K, decrypt ARMORED --key K, entropy TEXT|--file P
storage	tables DB, tail FILE [-n N], filter SRC DST --where EXPR, explain DB SQL [--param V], upsert DB TABLE ROW_JSON --on COL…
automation	exec [--timeout S] [--check] -- CMD…, exists NAME, cron EXPR [--count N], notify TITLE MSG
batch	[--workers N] [--inputs FILE] [--timeout S] -- CMD… {} … — parallel per-line runs, exit 11 on any failure

Every command supports --json. Destructive or scanning operations require an explicit --yes; public-IP lookups default to the ipify provider and can be overridden with --provider or DARK_NET_PUBLIC_IP_PROVIDERS.

Library tour

Configuration

dark_net.Config is an immutable, hashable dataclass. Resolution order is explicit overrides → DARK_NET_* environment variables → a config file (dark-net.toml / .json) → safe defaults. Retrieve the process-wide instance with get_config(), replace it with set_config(), clear it with reset_config().

from dark_net.core.config import Config, get_config, CONFIG_FIELDS

Config(timeout=5.0, verify_tls=True)  # validated on construction
Config.from_env()  # env only
Config.load(path="dark-net.toml")  # requires that file (else ConfigError)
Config().with_overrides(timeout=2.0)  # derived copy

for field in CONFIG_FIELDS:
    print(f"DARK_NET_{field.env:22} {field.kind:6} {field.help}")

Configurable keys: log_level, log_format, timeout, retries, backoff, verify_tls, allow_private_hosts, user_agent, max_response_bytes, max_redirects, cache_dir, public_ip_providers, exclude_dirs, consent.

Errors and exit codes

All exceptions derive from DarkNetError (message, code, details):

ConfigError, ValidationError, ParseError, SerializationError, NotFoundError, FileOperationError, PermissionDeniedError, PathSecurityError, NetworkError, ProtocolError, TimeoutError_, RateLimitError, SSRFBlockedError, SecurityError, IntegrityError, ConsentRequiredError, DependencyMissingError, BackupError, CacheError, CommandError, DataError, SystemError_.

from dark_net import ExitCode, ValidationError

try:
    ...
except ValidationError as exc:
    print(exc.message, exc.details)  # e.g. "invalid IP address" {'value': '…'}
    code = exc.exit_code  # 3 — the CLI reuses it as its process status

ExitCode: SUCCESS 0, GENERAL_ERROR 1, USAGE_ERROR 2, VALIDATION_ERROR 3, NOT_FOUND 4, PERMISSION_DENIED 5, NETWORK_ERROR 6, TIMEOUT 7, SECURITY_ERROR 8, CONSENT_REQUIRED 9, DEPENDENCY_MISSING 10, CHECK_FAILED 11.

System

from dark_net.system import (
    get_platform_info,
    get_cpu_info,
    get_memory_info,
    get_disk_usage,
    get_hardware_info,
    get_system_info,
    list_processes,
    get_process,
    EnvManager,
    is_sensitive_key,
)

get_system_info(include_hardware=True)  # dict of platform/cpu/memory/disk[/hardware]
get_disk_usage("/tmp").percent_used

env = EnvManager({"APP_MODE": "test"}, base={})
env.get("APP_MODE")
env.get_bool("FLAG", False)
env.redacted_dict()

Platform, CPU, memory and disk helpers return dataclasses with to_dict() and degrade to None rather than raising when the OS cannot report a value.

Network

from dark_net.network import (
    parse_ip,
    classify_address,
    is_unsafe_target,
    parse_network,
    iter_hosts,
    resolve_hostname,
    resolve_host,
    reverse_lookup,
    parse_url,
    validate_url,
    resolve_and_validate,
    sanitize_url_for_logging,
    HTTPClient,
    http_get,
    get_public_ip,
    check_port,
    measure_latency,
    list_interfaces,
    get_primary_ip,
)

parse_ip("2001:db8::1").version
classify_address("10.0.0.1")  # 'private'
is_unsafe_target("169.254.169.254")  # True (cloud metadata endpoint)

resolve_hostname("example.com")  # DNSResult(resolved=…, addresses=…)
resolve_host("example.com")  # raises NetworkError when it fails

client = HTTPClient(allow_private=True, timeout=5)  # opt in to private targets
response = client.get("http://127.0.0.1:8080/json")  # needs a server on that port
response.json()

get_public_ip(providers=["ipify"])  # PublicIPResult(found=…, error=…)

HTTPClient follows redirects manually (re-validating each hop, stripping Authorization/Cookie cross-origin), caps response bodies, honours a total deadline across hops, and retries only idempotent verbs. KNOWN_PROVIDERS is opt-in by design: ipify, ifconfig.co, icanhazip, ipinfo.

Port scanning is deliberately constrained — scan_ports/check_port require acknowledge_authorization=True, validate every port and cap the port budget.

Files

from dark_net.files import (
    read_text,
    write_text,
    write_json,
    read_json,
    load_csv,
    write_csv,
    copy_file,
    copy_tree,
    move_path,
    delete_tree,
    list_directory,
    hash_file,
    compute_hashes,
    find_duplicates,
    file_integrity_manifest,
    verify_integrity_manifest,
    create_zip,
    extract_zip,
    create_backup,
    verify_backup,
    restore_backup,
)

write_text("out/notes.txt", "hello\n", atomic=True, overwrite=True)
hash_file("out/notes.txt", algorithm="sha256")
find_duplicates("src")  # two-phase: size, then hash
manifest = file_integrity_manifest("src/")
verify_integrity_manifest("src/", manifest)  # {'modified': [], 'missing': [], 'unexpected': []}

# backup/verify/restore need an existing source directory and explicit consent
create_backup("docs/", "docs-backup.zip", consent=True)
verify_backup("docs-backup.zip").valid
restore_backup("docs-backup.zip", "restored/", consent=True)  # → restored/docs/…

Writes are atomic by default (temp file + os.replace). extract_zip blocks zip-slip, absolute paths and symlink members. delete_tree, create_backup, restore_backup and prune_backups all require consent=True.

Security

from dark_net.security import (
    redact,
    redact_dict,
    find_secrets,
    contains_secret,
    generate_password,
    estimate_password_strength,
    password_entropy_bits,
    generate_token,
    generate_passphrase,
    generate_api_key,
    hash_sha256,
    compare_digest,
    hmac_sign,
    verify_hmac,
    derive_key,
    audit_configuration,
    audit_environment,
    summarize_findings,
    compute_checksum,
    verify_file_checksum,
)

redact("Authorization: Bearer abc.def")  # Authorization: Bearer ***
estimate_password_strength("correct horse …").score
audit_names = [f.check for f in audit_configuration().findings]

Passwords are generated with secrets, strength is scored from length/entropy/ pattern heuristics, and audits return SecurityAuditReport objects with a 0–100 score, passed, by_severity and to_dict().

Text

from dark_net.text import (
    slugify,
    to_case,
    text_stats,
    reading_level,
    word_frequencies,
    truncate_words,
    template,
    strip_ansi,
    extract_urls,
    extract_emails,
    base64_encode,
    base64_decode,
    detect_encoding,
    to_ascii,
)

slugify("Hello, World!")  # 'hello-world'
to_case("myVarName", "snake")  # 'my_var_name'
text_stats("One two. Three.").to_dict()

Dates and times

from dark_net.datetime_utils import (
    now_utc,
    parse_date,
    format_date,
    format_duration,
    date_difference,
    business_days_between,
    start_of_day,
    end_of_day,
    date_range,
    to_timezone,
)

format_duration(3725)  # '1h 2m 5s'
date_difference("2024-01-01", "2024-03-01").days  # 60
business_days_between("2024-01-01", "2024-01-31")

Developer tooling

from dark_net.developer import (
    configure_logging,
    get_logger,
    JSONFormatter,
    RedactionFilter,
    RetryPolicy,
    retry,
    retry_call,
    async_retry,
    TTLCache,
    LRUCache,
    DiskCache,
    cached,
    profile,
    profile_call,
    timed,
    timed_block,
    debug_repr,
    to_json,
    from_json,
    Schema,
    Field,
    run_async,
)

configure_logging("DEBUG", json_format=True)


def fetch() -> str:
    return "payload"  # stand-in for a flaky network call


policy = RetryPolicy(attempts=3, base_delay=0.2, retry_on=(ConnectionError,))
result = retry_call(fetch, policy=policy)  # retries transient failures only

cache = TTLCache(maxsize=128, ttl=30)
profile_result = profile_call(fetch)  # (result, ProfileResult)

Web and HTTP

from dark_net.web import (
    JSONAPIClient,
    AsyncHTTPClient,
    parse_json_response,
    parse_pagination,
    extract_field,
    unwrap_api_error,
    build_url,
    redact_url,
)

client = JSONAPIClient("https://api.example.com", retries=2)  # doctest: +SKIP
response = client.get("/v1/items", params={"page": 2})  # doctest: +SKIP
response.ok, response.field("data.0.id"), response.pagination.has_next  # doctest: +SKIP


async def fetch_async() -> object:
    async with AsyncHTTPClient(timeout=10) as http:  # doctest: +SKIP
        return await http.get_json("https://api.example.com/v1/items")

AsyncHTTPClient speaks HTTP/1.1 over asyncio streams with the same SSRF and TLS rules; chunked responses are rejected rather than mis-parsed.

Data

from dark_net.data import (
    deep_merge,
    flatten,
    pick,
    pluck,
    group_by,
    sort_by,
    dedupe,
    filter_records,
    select_columns,
    to_records,
    chunk,
    transpose_records,
)

deep_merge({"a": {"b": 1}}, {"a": {"c": 2}})  # {'a': {'b': 1, 'c': 2}}
group_by([{"k": "a"}, {"k": "b"}], key=lambda r: r["k"])  # {'a': [...], 'b': [...]}
sort_by([{"n": 2}, {"n": 1}], "n")

Crypto

from dark_net import decrypt_text, encrypt_text, generate_key, shannon_entropy

key = generate_key()  # 32 CSPRNG bytes
armored = encrypt_text("payload", key)  # base64, authenticated
assert decrypt_text(armored, key) == "payload"
print(round(shannon_entropy(armored.encode()), 2))  # ~7.9 bits/byte

Also: digest/stream_digest, hmac_digest, hkdf/scrypt_derive, JWT encode/decode, base32/58/62/64/85 codecs, looks_random.

Storage

from dark_net import SQLiteStore, compile_filter, tail_jsonl, upsert

with SQLiteStore("app.db") as store:
    store.execute("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT)")
    upsert(store, "users", {"id": 1, "name": "ada"}, ["id"])
    print(store.query("SELECT * FROM users"))

print(tail_jsonl("events.jsonl", 5))  # last records, no full scan
predicate = compile_filter("level=error and retries>=3")

Concurrency

from dark_net import Pipeline, TokenBucket, parallel_map

print(parallel_map(abs, [-1, 2, -3], workers=3))  # [1, 2, 3]
print(Pipeline().map(lambda x: x * x).filter(lambda x: x % 2 == 0).run(range(10)))  # [0, 4, 16, 36, 64]

limiter = TokenBucket(rate=10, capacity=5)  # try_acquire / acquire

Automation

from dark_net import FileLock, exec_command, parse_cron, working_directory

result = exec_command(["echo", "hello"], check=True)
print(result.stdout.strip(), result.returncode)

with working_directory("/tmp"):
    print("scoped cwd")

with FileLock("/tmp/app.lock", timeout=0.5):
    print("single instance")

print(parse_cron("*/15 * * * *").next_after())  # next quarter-hour

Security posture

• SSRF protection — every outbound request (including each redirect hop) is resolved and checked. Loopback, RFC 1918, CGNAT, link-local (cloud metadata), multicast, reserved and unspecified ranges are blocked unless allow_private=True is set explicitly. Link-local and cloud metadata endpoints stay blocked even then.
• TLS — verification is on by default, TLS ≥ 1.2, and can only be disabled by passing verify_tls=False; no environment variable or query parameter can weaken it silently at request time.
• Consent gates — destructive filesystem operations (recursive deletes, backups, restores, pruning) require consent=True / --yes and otherwise raise ConsentRequiredError.
• Redaction — logs, CLI output and redact_dict/list_env mask credential- looking keys and values. --no-redact exists for debugging only.
• Resource limits — bounded response sizes, bounded archive expansion, bounded traversal depth, bounded process/port counts.
• No shell — subprocesses run with argv lists, shell=False, an allow-list of executables and explicit timeouts.
• Secret generation — secrets module only; comparisons use hmac.compare_digest.

Report vulnerabilities per SECURITY.md. Please do not open a public issue for a security problem.

Optional extras

Extra	Provides	Without it
system (psutil)	process lists, per-interface stats	/proc + socket fallbacks
yaml (PyYAML)	load_yaml / dump_yaml	DependencyMissingError with a hint
async (httpx)	convenience for async work	AsyncHTTPClient is stdlib-only
security (pip-audit)	authoritative vulnerability audit	PyPI "outdated" check
toml (tomli)	TOML config on Python 3.10	JSON config only

Development

python -m pip install -e ".[dev]"

ruff check . && ruff format --check .     # lint + formatting
mypy src/                                 # strict type-check
python -m pytest -q                       # full suite
python -m pytest -q --cov=dark_net        # coverage (fails under 90 %)
python -m pytest -q -m "not integration"  # offline subset
python -m build && twine check dist/*     # sdist + wheel sanity check

Markers: integration (filesystem/loopback/environment), security, slow, cli (spawns a subprocess). The default suite runs offline.

Documentation

• docs/index.md — documentation map
• docs/quickstart.md — five-minute tour
• docs/cli.md — every command and flag
• docs/library.md — module-by-module API guide
• docs/api/ — per-package references
• docs/security.md — threat model and guarantees
• docs/architecture.md — layering, module map, design rules
• docs/faq.md, docs/troubleshooting.md
• examples/ — runnable scripts
• scripts/ — check.sh, smoke_test.sh, build_dist.sh
• CHANGELOG.md — release history

License

MIT — see LICENSE.