2xppg 0.1.0

2xPPg (Peer to Peer Playground): dead-simple Python P2P building blocks.

2xPPg (Peer to Peer Playground)

By justharsiz

2xPPg means 2x "P" + "Pg":

• Peer + Peer = two equal participants
• Playground = a place to experiment and build

So this package is a Peer to Peer Playground for Python.

---

What is this?

2xPPg is a beginner-friendly Python toolkit for building peer-to-peer software without forcing you to become a networking expert first.

Think of it like pyautogui, but for p2p ideas:

• simple API
• quick feedback
• easy first success
• still flexible enough for advanced systems

You focus on:

• what your app does
• what data peers exchange
• how to react to incoming updates

2xPPg handles:

• socket setup
• peer handshake
• message framing
• peer loop and dispatch
• basic replicated state sync
• optional UPnP port mapping attempts

---

What you can build with it

1) Peer-to-peer messaging

One user sends text/data directly to another peer (or broadcasts to many).

2) Peer-to-peer file sharing (torrent-like foundations)

Peers host/share files. This package gives easy primitives for sending and receiving files. You can build chunk scheduling, swarm logic, and integrity verification on top.

3) Peer-managed shared state (blockchain-like patterns)

Each peer stores a shared state map and can periodically sync snapshots. You can layer consensus, signatures, and block rules above this.

4) Other data transfer patterns

Sensor data streaming, collaborative docs, game states, local mesh experiments, and more.

---

Important reality check (very important)

No networking library can magically guarantee universal inbound connectivity without any helper path.

In real life:

• Some peers can connect directly.
• Some need NAT traversal tricks.
• Some still need relay-style forwarding through another reachable peer.

2xPPg gives you:

1. Port mapping path (UPnP, easiest when available)
2. Direct peer path (if network allows)
3. Relay-ready design (you can forward through peers; no single mandatory central server)

So yes, you can avoid depending on one central always-on server, but practical networks still require traversal strategy.

---

Installation

pip install 2xppg

For local development:

python -m venv .venv
.venv\Scripts\activate
pip install -e .

---

60-second mental model

Imagine a group voice call:

• each person has a name (peer_id)
• each person can join someone (join(host, port))
• each person can send messages (send_text, send)
• each person can react to message types (on("text", handler))

Now replace voice with Python dict payloads.

That is 2xPPg.

---

Quick start: your first p2p app

Create chat_node.py:

import asyncio
from xppg import Playground


async def main():
    pg = Playground(app_name="my-first-chat", port=7337)
    nat = await pg.start(open_port=True)
    print("I am:", pg.peer_id)
    if nat:
        print("NAT result:", nat)

    pg.on("text", lambda src, kind, payload: print(f"[{src}] {payload['text']}"))

    # If this node should join another node:
    # await pg.join("192.168.1.42", 7337)

    while True:
        text = await asyncio.to_thread(input, "> ")
        if text.strip().lower() in {"quit", "exit"}:
            break
        await pg.send_text(text)

    await pg.stop()


if __name__ == "__main__":
    asyncio.run(main())

Run on two machines in the same LAN (or otherwise reachable):

python chat_node.py

On one machine, uncomment join() and point it at the first.

---

API overview (beginner language)

Playground(app_name, host="0.0.0.0", port=7337)

Creates one p2p node.

• app_name: your app/protocol namespace
• host: local bind host
• port: local listening port
• peer_id: auto-generated identity string
• state: built-in replicated key/value state (SharedState)

await pg.start(open_port=True)

Starts listening server.

• if open_port=True, tries UPnP port mapping
• returns NAT result object or None

await pg.join(host, port)

Connects to another peer and performs handshake.

pg.on(kind, handler)

Registers a handler for a message type. Handler signature:

def handler(src_peer_id: str, kind: str, payload: dict) -> None:
    ...

await pg.send(kind, payload, to=None)

Send custom message type to one peer (to="peer_id") or broadcast (to=None).

await pg.send_text(text, to=None)

Convenience for text messages.

await pg.share_file(path, to=None)

Sends a file payload. (In this alpha release, this is whole-file base64 transport. For production-scale torrents, use chunking + verification strategy.)

await pg.request_state(to=None) / await pg.push_state(to=None)

State sync helpers.

---

Shared state for "everyone manages something"

The SharedState class is the simplest version of a replicated map:

• keys
• values
• versions
• optional merge function for tie versions

Example:

from xppg import SharedState

state = SharedState()
state.set("height", 1)
state.set("height", 2)
print(state.get("height"))  # 2

When peers exchange state snapshots (state_update messages), newer versions win by default.

This is enough to prototype:

• distributed counters
• replicated settings
• toy blockchain headers/metadata

For real blockchain logic, add:

• signed blocks/transactions
• deterministic validation rules
• consensus/fork-choice
• anti-replay/anti-spam rules

---

File sharing model (torrent-like direction)

Current helper:

• share_file(path) sends one encoded blob to target/broadcast

To build torrent-like transfer on top, extend into:

1. split file into chunks
2. hash each chunk
3. share metadata manifest
4. request missing chunks from many peers
5. verify chunks before assembling final file

2xPPg is the playground foundation for this.

---

NAT, port forwarding, and "no central server"

Why this matters

Most people are behind routers and NAT. Inbound connections can fail unless:

• router forwards a port
• both peers hole-punch successfully
• or traffic relays through reachable peers

What 2xPPg currently includes

• UPnP mapping helper (try_upnp_map) for easy cases
• peer architecture that does not force one central coordinator

Recommended architecture for hard NAT environments

Use multi-hop peer relay:

• any publicly reachable peer can relay encrypted packets
• clients can rotate relay peers
• no single relay is required globally

This keeps your system decentralized in operation, even when direct routes are blocked.

---

Example patterns

Pattern A: direct messaging app

• start node
• join known peer(s)
• send text/data events

Pattern B: collaborative state app

• each peer keeps SharedState
• periodic push_state
• on state update, render local UI or trigger actions

Pattern C: p2p currency prototype

• represent wallet balances/UTXO entries in replicated structures
• broadcast transactions as signed messages
• update state by deterministic rules
• add block production/finality module above it

---

Beginner analogies

• Peer: a person in a group chat
• Message kind: topic label ("chat", "file", "tx", "state_update")
• Payload: envelope contents
• Handshake: saying "hello, this is me"
• NAT: apartment security desk that blocks unknown visitors
• Port forwarding: adding your name to allowed visitor list
• Relay peer: trusted friend who forwards messages for you

If networking feels hard, remember: You are designing message flow first. Transport details are implementation layers.

---

Security notes (read this before production)

This alpha package prioritizes simplicity and education. For production:

• add payload encryption (Noise/TLS or libsodium patterns)
• sign messages
• rate limit and ban abusive peers
• chunk large files with checksums
• sandbox untrusted content
• add peer reputation/allow-lists
• validate all input strictly

---

Full mini project: state + messaging

import asyncio
from xppg import Playground


async def main():
    pg = Playground("ledger-playground", port=7444)
    await pg.start()

    def on_text(src, _kind, payload):
        print(f"[msg:{src}] {payload['text']}")

    pg.on("text", on_text)

    # toy ledger state
    pg.state.set("alice", 100)
    pg.state.set("bob", 50)

    # join a seed peer (if you have one)
    # await pg.join("SEED_IP", 7444)

    # periodically sync state snapshots
    async def sync_loop():
        while True:
            await pg.push_state()
            await asyncio.sleep(5)

    task = asyncio.create_task(sync_loop())

    try:
        while True:
            cmd = await asyncio.to_thread(input, "cmd> ")
            if cmd == "quit":
                break
            if cmd.startswith("pay "):
                _, src, dst, amt = cmd.split()
                amt = int(amt)
                pg.state.set(src, pg.state.get(src, 0) - amt)
                pg.state.set(dst, pg.state.get(dst, 0) + amt)
                await pg.push_state()
                continue
            await pg.send_text(cmd)
    finally:
        task.cancel()
        await pg.stop()


if __name__ == "__main__":
    asyncio.run(main())

---

Package layout

• src/xppg/playground.py - main high-level p2p API
• src/xppg/state.py - replicated shared state map
• src/xppg/nat.py - UPnP mapping helper

---

Exporting to PyPI

1) Build

python -m pip install --upgrade build twine
python -m build

2) Check artifacts

python -m twine check dist/*

3) Upload to TestPyPI first (recommended)

python -m twine upload --repository testpypi dist/*

4) Upload to PyPI

python -m twine upload dist/*

5) Install to verify

pip install 2xppg

---

Roadmap (next features)

• UDP hole punching helper module
• pluggable relay mesh utilities
• chunked file transfer protocol with retries
• optional transport encryption helpers
• discovery module (LAN + DHT adapter hooks)
• protocol versioning and capability negotiation

---

License

MIT