Skip to content

API Reference

This section documents the API of p2p_copy. Only the functions run_relay, send and receive are intended for users, others are mainly information for developers. For module structure, see Module Layout. Generated from source code docstrings.

 

p2p_copy_server

run_relay async 

run_relay(host, port, use_tls=True, certfile=None, keyfile=None)

Run the WebSocket relay server for pairing and forwarding client connections.

This command starts a relay server that listens on the specified host/interface and port, optionally secured with TLS (recommended for production). The server pairs sender and receiver clients based on matching passphrase hashes, then forwards bidirectional data streams without storing content. It handles exactly one sender and one receiver per code hash, rejecting duplicates. Use for secure, firewall-friendly (port 443) P2P transfers.

Parameters:

Name Type Description Default
host str

Host to bind to.

 required
port int

Port to bind to.

 required
use_tls bool

Whether to use TLS. Default is True.

 True
certfile str

Path to TLS certificate file.

 None
keyfile str

Path to TLS key file.

 None

Raises:

Type Description
RuntimeError

If TLS is requested but certfile or keyfile is missing.
Source code in src/p2p_copy_server/relay.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
async def run_relay(host: str, port: int,
                    use_tls: bool = True,
                    certfile: Optional[str] = None,
                    keyfile: Optional[str] = None) -> None:
    """
    Run the WebSocket relay server for pairing and forwarding client connections.

    This command starts a relay server that listens on the specified host/interface and port,
    optionally secured with TLS (recommended for production). The server pairs
    sender and receiver clients based on matching passphrase hashes, then forwards
    bidirectional data streams without storing content. It handles exactly one
    sender and one receiver per code hash, rejecting duplicates. Use for secure,
    firewall-friendly (port 443) P2P transfers.

    Parameters
    ----------
    host : str
        Host to bind to.
    port : int
        Port to bind to.
    use_tls : bool, optional
        Whether to use TLS. Default is True.
    certfile : str, optional
        Path to TLS certificate file.
    keyfile : str, optional
        Path to TLS key file.

    Raises
    ------
    RuntimeError
        If TLS is requested but certfile or keyfile is missing.
    """
    ssl_ctx = None
    if use_tls:
        if not certfile or not keyfile:
            raise RuntimeError("TLS requested but certfile/keyfile missing")
        ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ssl_ctx.load_cert_chain(certfile, keyfile)

    scheme = "wss" if ssl_ctx else "ws"
    print(f"\nRelay listening on {scheme}://{host}:{port}")

    if host != "localhost":
        use_production_logger()

    async with serve(_handle, host, port, max_size=2**21, ssl=ssl_ctx, compression=None):
        await asyncio.Future()  # run forever

 

p2p_copy.api

send async 

send(server, code, files, *, encrypt=False, compress=CompressMode.auto, resume=False)

Send one or more files or directories to a paired receiver via the relay server.

This command connects to the specified WebSocket relay server, authenticates using the shared passphrase (hashed for pairing), and streams the provided files/directories in chunks to the receiver. Supports directories by recursively including all files in alphabetical order. Optional end-to-end encryption (AES-GCM) and compression (Zstandard, auto-detected per file) can be enabled. If resume is enabled, it coordinates with the receiver to skip complete files or append to partial ones based on chained checksum verification. By prefixing the pairing code with n={N}=n, e.g., n=4=nMySecret, file transfers will be done over N connections.

Parameters:

Name Type Description Default
server str

The WebSocket server URL (ws:// or wss://).

 required
code str

The shared passphrase/code for pairing. Prefix with n={N}=n to use N parallel connections. Receiver needs to use the exact same code.

 required
files List[str]

List of files and/or directories to send.

 required
encrypt bool

Enable end-to-end encryption. Default is False. Receiver needs to use the same setting.

 False
compress CompressMode

Compression mode. Default is 'auto'.

 auto
resume bool

Enable resume of partial transfers. Default is False. If True, attempt to skip identical files and append incomplete files based on receiver feedback.

 False

Returns:

Type Description
int

Exit code: 0 on success, non-zero on error.
Notes
  • Supports resuming by comparing checksums of partial files.
  • Uses chunked streaming for large files.
Source code in src/p2p_copy/api.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def send(server: str, code: str, files: List[str],
               *, encrypt: bool = False,
               compress: CompressMode = CompressMode.auto,
               resume: bool = False) -> int:
    """
    Send one or more files or directories to a paired receiver via the relay server.

    This command connects to the specified WebSocket relay server, authenticates using
    the shared passphrase (hashed for pairing), and streams the provided files/directories
    in chunks to the receiver. Supports directories by recursively including all files
    in alphabetical order. Optional end-to-end encryption (AES-GCM) and compression
    (Zstandard, auto-detected per file) can be enabled. If resume is enabled, it
    coordinates with the receiver to skip complete files or append to partial ones
    based on chained checksum verification. By prefixing the pairing code with n={N}=n,
     e.g., n=4=nMySecret, file transfers will be done over N connections.

    Parameters
    ----------
    server : str
        The WebSocket server URL (ws:// or wss://).
    code : str
        The shared passphrase/code for pairing. Prefix with n={N}=n to use N parallel connections.
        Receiver needs to use the exact same code.
    files : List[str]
        List of files and/or directories to send.
    encrypt : bool, optional
        Enable end-to-end encryption. Default is False.
        Receiver needs to use the same setting.
    compress : CompressMode, optional
        Compression mode. Default is 'auto'.
    resume : bool, optional
        Enable resume of partial transfers. Default is False.
        If True, attempt to skip identical files and append
        incomplete files based on receiver feedback.

    Returns
    -------
    int
        Exit code: 0 on success, non-zero on error.

    Notes
    -----
    - Supports resuming by comparing checksums of partial files.
    - Uses chunked streaming for large files.
    """

    # get amount of parallel connections
    n = 1
    if code.startswith("n=") and "=n" in code:
        _n = code[2:code.index("=n")]
        if _n.isdigit():
            n = max(1, int(_n))

    # split the files to send in n groups of files with about equal size (in terms of bytes per group)
    resolved_file_list: List[List[Tuple[Path, Path, int]]] = iter_manifest_entries(files, n)
    if not any(resolved_file_list):
        print("[p2p_copy] send(): no legal files where passed")
        return 3

    if n == 1:
        return await _inner_send(server, code, file_group=resolved_file_list[0],
                                 encrypt=encrypt,
                                 compress=compress,
                                 resume=resume)
    else:
        send_tasks = []
        for i in range(n):
            file_group = resolved_file_list[i]
            send_tasks.append(asyncio.create_task(
                _inner_send(server, str(i) + code, file_group=file_group,
                            encrypt=encrypt, compress=compress,
                            resume=resume)))

        return_codes = await asyncio.gather(*send_tasks, return_exceptions=True)
        return 3 if any(return_codes) else 0

receive async 

receive(server, code, *, encrypt=False, out=None)

Receive files from a paired sender via the relay server and write to the output directory.

This command connects to the relay server, pairs using the shared passphrase hash, and receives a manifest of incoming files/directories. Files are written to the output directory, preserving relative paths from the manifest. Supports optional end-to-end decryption (matching sender's encryption) and decompression. If the sender requests resume, this receiver reports existing file states (via checksums) to enable skipping or appending. By prefixing the pairing code with n={N}=n, e.g., n=4=nMySecret, file transfers will be done over N connections.

Parameters:

Name Type Description Default
server str

The WebSocket server URL (ws:// or wss://).

 required
code str

The shared passphrase/code for pairing. Prefix with n={N}=n to use N parallel connections. Sender needs to use the exact same code.

 required
encrypt bool

Enable end-to-end encryption. Default is False. Sender needs to use the same setting.

 False
out str

Output directory. Default is current directory.

 None

Returns:

Type Description
int

Exit code: 0 on success, non-zero on error.
Notes
  • Supports resume if sender requests it.
  • Writes files to the output directory, preserving relative paths.
  • Info on whether to resume and compress is received from the sender
Source code in src/p2p_copy/api.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
async def receive(server: str, code: str,
                  *, encrypt: bool = False,
                  out: Optional[str] = None) -> int:
    """
    Receive files from a paired sender via the relay server and write to the output directory.

    This command connects to the relay server, pairs using the shared passphrase hash,
    and receives a manifest of incoming files/directories. Files are written to the
    output directory, preserving relative paths from the manifest. Supports optional
    end-to-end decryption (matching sender's encryption) and decompression. If the
    sender requests resume, this receiver reports existing file states (via checksums)
    to enable skipping or appending. By prefixing the pairing code with n={N}=n,
     e.g., n=4=nMySecret, file transfers will be done over N connections.

    Parameters
    ----------
    server : str
        The WebSocket server URL (ws:// or wss://).
    code : str
        The shared passphrase/code for pairing. Prefix with n={N}=n to use N parallel connections.
        Sender needs to use the exact same code.
    encrypt : bool, optional
        Enable end-to-end encryption. Default is False.
        Sender needs to use the same setting.
    out : str, optional
        Output directory. Default is current directory.

    Returns
    -------
    int
        Exit code: 0 on success, non-zero on error.

    Notes
    -----
    - Supports resume if sender requests it.
    - Writes files to the output directory, preserving relative paths.
    - Info on whether to resume and compress is received from the sender
    """

    # ensure out directory exists
    out_dir = Path(out or ".")
    ensure_dir(out_dir)

    # get amount of parallel connections
    n = 1
    if code.startswith("n=") and "=n" in code:
        _n = code[2:code.index("=n")]
        if _n.isdigit():
            n = max(1, int(_n))

    if n == 1:
        return await _inner_receive(server, code, encrypt=encrypt, out_dir=out_dir)

    else:
        receive_tasks = []
        for i in range(n):
            # create task to receive that runs in parallel
            # use a modified code for each
            receive_tasks.append(asyncio.create_task(
                _inner_receive(server, code= str(i) + code,
                               encrypt=encrypt, out_dir=out_dir)))

        return_codes = await asyncio.gather(*receive_tasks, return_exceptions=True)
        return 4 if any(return_codes) else 0

 

p2p_copy.security

SecurityHandler

Handle security operations like hashing, encryption, and decryption for transfers.

Parameters:

Name Type Description Default
code str

The shared passphrase/code.

 required
encrypt bool

Whether to enable end-to-end encryption.

 required
Source code in src/p2p_copy/security.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class SecurityHandler:
    """
    Handle security operations like hashing, encryption, and decryption for transfers.

    Parameters
    ----------
    code : str
        The shared passphrase/code.
    encrypt : bool
        Whether to enable end-to-end encryption.
    """

    def __init__(self, code: str, encrypt: bool):
        self.encrypt = encrypt
        if self.encrypt:
            import_optional_security_libs()
            self.code_hash = _get_argon2_hash(code, b"code_hash used for hello-match")
            self.nonce_hasher = ChainedChecksum()
            self.cipher = AESGCM(_get_argon2_hash(code, b"cipher used for E2E-encryption"))
        else:
            self.code_hash = hashlib.sha256(code.encode()).digest()

    def encrypt_chunk(self, chunk: bytes) -> bytes:
        """
        Encrypt a chunk if encryption is enabled.

        Parameters
        ----------
        chunk : bytes
            The chunk to encrypt.

        Returns
        -------
        bytes
            The encrypted chunk, or original if not encrypted.
        """
        if self.encrypt:
            return self.cipher.encrypt(self.nonce_hasher.next_hash(), chunk, None)
        return chunk

    def decrypt_chunk(self, chunk: bytes) -> bytes:
        """
        Decrypt a chunk if encryption is enabled.

        Parameters
        ----------
        chunk : bytes
            The chunk to decrypt.

        Returns
        -------
        bytes
            The decrypted chunk, or original if not encrypted.
        """
        if self.encrypt:
            return self.cipher.decrypt(self.nonce_hasher.next_hash(), chunk, None)
        return chunk

    def build_encrypted_manifest(self, manifest: str) -> str:
        """
        Build an encrypted manifest for secure transmission.

        Parameters
        ----------
        manifest : str
            The plaintext manifest JSON.

        Returns
        -------
        str
            The JSON-serialized EncryptedManifest.
        """
        start_nonce = os.urandom(32)
        self.nonce_hasher.next_hash(start_nonce)
        enc_manifest = self.encrypt_chunk(manifest.encode())
        return EncryptedManifest(
            type="enc_manifest",
            nonce=start_nonce.hex(),
            hidden_manifest=enc_manifest.hex()
        ).to_json()

encrypt_chunk 

encrypt_chunk(chunk)

Encrypt a chunk if encryption is enabled.

Parameters:

Name Type Description Default
chunk bytes

The chunk to encrypt.

 required

Returns:

Type Description
bytes

The encrypted chunk, or original if not encrypted.
Source code in src/p2p_copy/security.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def encrypt_chunk(self, chunk: bytes) -> bytes:
    """
    Encrypt a chunk if encryption is enabled.

    Parameters
    ----------
    chunk : bytes
        The chunk to encrypt.

    Returns
    -------
    bytes
        The encrypted chunk, or original if not encrypted.
    """
    if self.encrypt:
        return self.cipher.encrypt(self.nonce_hasher.next_hash(), chunk, None)
    return chunk

decrypt_chunk 

decrypt_chunk(chunk)

Decrypt a chunk if encryption is enabled.

Parameters:

Name Type Description Default
chunk bytes

The chunk to decrypt.

 required

Returns:

Type Description
bytes

The decrypted chunk, or original if not encrypted.
Source code in src/p2p_copy/security.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def decrypt_chunk(self, chunk: bytes) -> bytes:
    """
    Decrypt a chunk if encryption is enabled.

    Parameters
    ----------
    chunk : bytes
        The chunk to decrypt.

    Returns
    -------
    bytes
        The decrypted chunk, or original if not encrypted.
    """
    if self.encrypt:
        return self.cipher.decrypt(self.nonce_hasher.next_hash(), chunk, None)
    return chunk

build_encrypted_manifest 

build_encrypted_manifest(manifest)

Build an encrypted manifest for secure transmission.

Parameters:

Name Type Description Default
manifest str

The plaintext manifest JSON.

 required

Returns:

Type Description
str

The JSON-serialized EncryptedManifest.
Source code in src/p2p_copy/security.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def build_encrypted_manifest(self, manifest: str) -> str:
    """
    Build an encrypted manifest for secure transmission.

    Parameters
    ----------
    manifest : str
        The plaintext manifest JSON.

    Returns
    -------
    str
        The JSON-serialized EncryptedManifest.
    """
    start_nonce = os.urandom(32)
    self.nonce_hasher.next_hash(start_nonce)
    enc_manifest = self.encrypt_chunk(manifest.encode())
    return EncryptedManifest(
        type="enc_manifest",
        nonce=start_nonce.hex(),
        hidden_manifest=enc_manifest.hex()
    ).to_json()

ChainedChecksum

Generate chained SHA-256 checksums over sequential payloads.

Parameters:

Name Type Description Default
seed bytes

Initial seed for the chain. Default is empty bytes.

 b''
Source code in src/p2p_copy/security.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class ChainedChecksum:
    """
    Generate chained SHA-256 checksums over sequential payloads.

    Parameters
    ----------
    seed : bytes, optional
        Initial seed for the chain. Default is empty bytes.
    """

    def __init__(self, seed: bytes = b"") -> None:
        self.prev_chain = seed

    def next_hash(self, payload: bytes = b"") -> bytes:
        """
        Compute the next hash in the chain: sha256(prev_chain || payload).

        Parameters
        ----------
        payload : bytes, optional
            Data to include in this hash. Default is empty.

        Returns
        -------
        bytes
            The 32-byte hash, which becomes the new prev_chain.
        """
        h = hashlib.sha256()
        h.update(self.prev_chain)
        h.update(payload)
        self.prev_chain = h.digest()
        return self.prev_chain

next_hash 

next_hash(payload=b'')

Compute the next hash in the chain: sha256(prev_chain || payload).

Parameters:

Name Type Description Default
payload bytes

Data to include in this hash. Default is empty.

 b''

Returns:

Type Description
bytes

The 32-byte hash, which becomes the new prev_chain.
Source code in src/p2p_copy/security.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def next_hash(self, payload: bytes = b"") -> bytes:
    """
    Compute the next hash in the chain: sha256(prev_chain || payload).

    Parameters
    ----------
    payload : bytes, optional
        Data to include in this hash. Default is empty.

    Returns
    -------
    bytes
        The 32-byte hash, which becomes the new prev_chain.
    """
    h = hashlib.sha256()
    h.update(self.prev_chain)
    h.update(payload)
    self.prev_chain = h.digest()
    return self.prev_chain

import_optional_security_libs 

import_optional_security_libs()

Import optional security libraries (argon2-cffi, cryptography) if encryption is used.

Source code in src/p2p_copy/security.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
def import_optional_security_libs():
    """
    Import optional security libraries (argon2-cffi, cryptography) if encryption is used.
    """
    global hash_secret_raw, Type, AESGCM
    try:
        # security libs are needed if encryption is used
        from argon2.low_level import hash_secret_raw, Type
        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    except ModuleNotFoundError as E:
        raise ModuleNotFoundError(
            E.msg + '\nTo use encryption optional security libs are needed (pip install p2p-copy[security])')

 

p2p_copy.compressor

CompressMode

Bases: str, Enum

Enumeration of compression modes.

Source code in src/p2p_copy/compressor.py 
 7
 8
 9
10
11
12
13
14
class CompressMode(str, Enum):
    """
    Enumeration of compression modes.
    """

    auto = "auto"
    on = "on"
    off = "off"

Compressor

Handle compression and decompression of chunks using Zstandard.

Parameters:

Name Type Description Default
mode CompressMode

Compression mode. Default is 'auto'.

 auto
Source code in src/p2p_copy/compressor.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class Compressor:
    """
    Handle compression and decompression of chunks using Zstandard.

    Parameters
    ----------
    mode : CompressMode, optional
        Compression mode. Default is 'auto'.
    """

    def __init__(self, mode: CompressMode = CompressMode.auto):
        self.mode = mode
        self.cctx: Optional[zstd.ZstdCompressor] = zstd.ZstdCompressor(level=3) if mode != CompressMode.off else None
        self.dctx: Optional[zstd.ZstdDecompressor] = None
        self.use_compression: bool = mode == CompressMode.on
        self.compression_type: str = "zstd" if mode == CompressMode.on else "none"

    async def determine_compression(self, first_chunk: bytes) -> bytes:
        """
        Determine if compression should be used based on the first chunk (auto mode).

        Parameters
        ----------
        first_chunk : bytes
            The first chunk of data.

        Returns
        -------
        bytes
            The (possibly compressed) first chunk.
        """

        if self.mode == CompressMode.off:
            return first_chunk

        else:
            compressed = self.cctx.compress(first_chunk)
            if self.mode == CompressMode.on:
                return compressed

            elif self.mode == CompressMode.auto:
                # Auto mode: test first chunk
                compression_ratio = len(compressed) / len(first_chunk) if first_chunk else 1.0
                self.use_compression = compression_ratio < 0.95  # Enable if compressed size < 95% of original
                self.compression_type = "zstd" if self.use_compression else "none"
                return compressed if self.use_compression else first_chunk

    def compress(self, chunk: bytes) -> bytes:
        """
        Compress a chunk if compression is enabled.

        Parameters
        ----------
        chunk : bytes
            The chunk to compress.

        Returns
        -------
        bytes
            The compressed or original chunk.
        """
        """Compress a chunk if compression is enabled."""
        if self.use_compression and self.cctx:
            return self.cctx.compress(chunk)
        return chunk

    def decompress(self, chunk: bytes) -> bytes:
        """
        Decompress a chunk if decompression is set up.

        Parameters
        ----------
        chunk : bytes
            The chunk to decompress.

        Returns
        -------
        bytes
            The decompressed or original chunk.
        """

        if self.dctx:
            return self.dctx.decompress(chunk)
        return chunk

    def set_decompression(self, compression_type: str):
        """
        Set up the decompressor based on the compression type.

        Parameters
        ----------
        compression_type : str
            The type of compression ('zstd' or 'none').
        """

        self.dctx = zstd.ZstdDecompressor() if compression_type == "zstd" else None

determine_compression async 

determine_compression(first_chunk)

Determine if compression should be used based on the first chunk (auto mode).

Parameters:

Name Type Description Default
first_chunk bytes

The first chunk of data.

 required

Returns:

Type Description
bytes

The (possibly compressed) first chunk.
Source code in src/p2p_copy/compressor.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
async def determine_compression(self, first_chunk: bytes) -> bytes:
    """
    Determine if compression should be used based on the first chunk (auto mode).

    Parameters
    ----------
    first_chunk : bytes
        The first chunk of data.

    Returns
    -------
    bytes
        The (possibly compressed) first chunk.
    """

    if self.mode == CompressMode.off:
        return first_chunk

    else:
        compressed = self.cctx.compress(first_chunk)
        if self.mode == CompressMode.on:
            return compressed

        elif self.mode == CompressMode.auto:
            # Auto mode: test first chunk
            compression_ratio = len(compressed) / len(first_chunk) if first_chunk else 1.0
            self.use_compression = compression_ratio < 0.95  # Enable if compressed size < 95% of original
            self.compression_type = "zstd" if self.use_compression else "none"
            return compressed if self.use_compression else first_chunk

compress 

compress(chunk)

Compress a chunk if compression is enabled.

Parameters:

Name Type Description Default
chunk bytes

The chunk to compress.

 required

Returns:

Type Description
bytes

The compressed or original chunk.
Source code in src/p2p_copy/compressor.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def compress(self, chunk: bytes) -> bytes:
    """
    Compress a chunk if compression is enabled.

    Parameters
    ----------
    chunk : bytes
        The chunk to compress.

    Returns
    -------
    bytes
        The compressed or original chunk.
    """
    """Compress a chunk if compression is enabled."""
    if self.use_compression and self.cctx:
        return self.cctx.compress(chunk)
    return chunk

decompress 

decompress(chunk)

Decompress a chunk if decompression is set up.

Parameters:

Name Type Description Default
chunk bytes

The chunk to decompress.

 required

Returns:

Type Description
bytes

The decompressed or original chunk.
Source code in src/p2p_copy/compressor.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def decompress(self, chunk: bytes) -> bytes:
    """
    Decompress a chunk if decompression is set up.

    Parameters
    ----------
    chunk : bytes
        The chunk to decompress.

    Returns
    -------
    bytes
        The decompressed or original chunk.
    """

    if self.dctx:
        return self.dctx.decompress(chunk)
    return chunk

set_decompression 

set_decompression(compression_type)

Set up the decompressor based on the compression type.

Parameters:

Name Type Description Default
compression_type str

The type of compression ('zstd' or 'none').

 required
Source code in src/p2p_copy/compressor.py 
102
103
104
105
106
107
108
109
110
111
112
def set_decompression(self, compression_type: str):
    """
    Set up the decompressor based on the compression type.

    Parameters
    ----------
    compression_type : str
        The type of compression ('zstd' or 'none').
    """

    self.dctx = zstd.ZstdDecompressor() if compression_type == "zstd" else None

 

p2p_copy.protocol

Hello dataclass

Hello message for connection initiation.

Parameters:

Name Type Description Default
type Literal['hello']

Message type.

 required
code_hash_hex str

Hex-encoded hash of the shared code.

 required
role Literal['sender', 'receiver']

The role of this client.

 required
Source code in src/p2p_copy/protocol.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass(frozen=True)
class Hello:
    """
    Hello message for connection initiation.

    Parameters
    ----------
    type : Literal["hello"]
        Message type.
    code_hash_hex : str
        Hex-encoded hash of the shared code.
    role : Literal["sender", "receiver"]
        The role of this client.
    """
    type: Literal["hello"]
    code_hash_hex: str
    role: Literal["sender", "receiver"]

    def to_json(self) -> str:
        return dumps({"type": "hello", "code_hash_hex": self.code_hash_hex, "role": self.role})

ManifestEntry dataclass

Entry in a file manifest.

Parameters:

Name Type Description Default
path str

Relative path of the file.

 required
size int

File size in bytes.

 required
Source code in src/p2p_copy/protocol.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass(frozen=True)
class ManifestEntry:
    """
    Entry in a file manifest.

    Parameters
    ----------
    path : str
        Relative path of the file.
    size : int
        File size in bytes.
    """
    path: str
    size: int

Manifest dataclass

Manifest of files to send.

Parameters:

Name Type Description Default
type Literal['manifest']

Message type.

 required
entries Sequence[ManifestEntry]

List of file entries.

 required
resume bool

Whether to enable resume. Default is False.

 False
Source code in src/p2p_copy/protocol.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@dataclass(frozen=True)
class Manifest:
    """
    Manifest of files to send.

    Parameters
    ----------
    type : Literal["manifest"]
        Message type.
    entries : Sequence[ManifestEntry]
        List of file entries.
    resume : bool, optional
        Whether to enable resume. Default is False.
    """
    type: Literal["manifest"]
    entries: Sequence[ManifestEntry]
    resume: bool = False

    def to_json(self) -> str:
        return dumps({
            "type": "manifest",
            "resume": self.resume,
            "entries": [asdict(e) for e in self.entries]
        })

EncryptedManifest dataclass

Encrypted manifest for secure transmission.

Parameters:

Name Type Description Default
type Literal['enc_manifest']

Message type.

 required
nonce str

Hex-encoded nonce that is used as random seed. Further nonces are based on this and used for encryption. Must be shared with receiver so it can decrypt accordingly.

 required
hidden_manifest str

Hex-encoded encrypted manifest.

 required
Source code in src/p2p_copy/protocol.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@dataclass(frozen=True)
class EncryptedManifest:
    """
    Encrypted manifest for secure transmission.

    Parameters
    ----------
    type : Literal["enc_manifest"]
        Message type.
    nonce : str
        Hex-encoded nonce that is used as random seed.
        Further nonces are based on this and used for encryption.
        Must be shared with receiver so it can decrypt accordingly.
    hidden_manifest : str
        Hex-encoded encrypted manifest.
    """
    type: Literal["enc_manifest"]
    nonce: str
    hidden_manifest: str

    def to_json(self) -> str:
        return dumps({
            "type": "enc_manifest",
            "nonce": self.nonce,
            "hidden_manifest": self.hidden_manifest
        })

ReceiverManifestEntry dataclass

Receiver's report of existing file state for resume.

Parameters:

Name Type Description Default
path str

Relative path.

 required
size int

Bytes already present.

 required
chain_hex str

Hex-encoded chained checksum up to 'size'.

 required
Source code in src/p2p_copy/protocol.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@dataclass(frozen=True)
class ReceiverManifestEntry:
    """
    Receiver's report of existing file state for resume.

    Parameters
    ----------
    path : str
        Relative path.
    size : int
        Bytes already present.
    chain_hex : str
        Hex-encoded chained checksum up to 'size'.
    """
    path: str
    size: int
    chain_hex: str

ReceiverManifest dataclass

Manifest from receiver reporting existing files.

Parameters:

Name Type Description Default
type Literal['receiver_manifest']

Message type.

 required
entries Sequence[ReceiverManifestEntry]

List of entries.

 required
Source code in src/p2p_copy/protocol.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@dataclass(frozen=True)
class ReceiverManifest:
    """
    Manifest from receiver reporting existing files.

    Parameters
    ----------
    type : Literal["receiver_manifest"]
        Message type.
    entries : Sequence[ReceiverManifestEntry]
        List of entries.
    """
    type: Literal["receiver_manifest"]
    entries: Sequence[ReceiverManifestEntry]

    def to_json(self) -> str:
        return dumps({
            "type": "receiver_manifest",
            "entries": [asdict(e) for e in self.entries]
        })

EncryptedReceiverManifest dataclass

Encrypted receiver manifest.

Parameters:

Name Type Description Default
type Literal['enc_receiver_manifest']

Message type.

 required
hidden_manifest str

Hex-encoded encrypted manifest.

 required
Source code in src/p2p_copy/protocol.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@dataclass(frozen=True)
class EncryptedReceiverManifest:
    """
    Encrypted receiver manifest.

    Parameters
    ----------
    type : Literal["enc_receiver_manifest"]
        Message type.
    hidden_manifest : str
        Hex-encoded encrypted manifest.
    """
    type: Literal["enc_receiver_manifest"]
    hidden_manifest: str

    def to_json(self) -> str:
        return dumps({
            "type": "enc_receiver_manifest",
            "hidden_manifest": self.hidden_manifest
        })

dumps 

dumps(msg)

JSON-dump a message with compact separators.

Parameters:

Name Type Description Default
msg Dict[str, Any]

The message to serialize.

 required

Returns:

Type Description
str

Compact JSON string.
Source code in src/p2p_copy/protocol.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def dumps(msg: Dict[str, Any]) -> str:
    """
    JSON-dump a message with compact separators.

    Parameters
    ----------
    msg : Dict[str, Any]
        The message to serialize.

    Returns
    -------
    str
        Compact JSON string.
    """
    return json.dumps(msg, separators=(",", ":"), ensure_ascii=False)

loads 

loads(s)

JSON-load a string into a dict.

Parameters:

Name Type Description Default
s str

JSON string.

 required

Returns:

Type Description
Dict[str, Any]

Parsed dictionary.
Source code in src/p2p_copy/protocol.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def loads(s: str) -> Dict[str, Any]:
    """
    JSON-load a string into a dict.

    Parameters
    ----------
    s : str
        JSON string.

    Returns
    -------
    Dict[str, Any]
        Parsed dictionary.
    """
    return json.loads(s)

file_begin 

file_begin(path, size, compression='none', append_from=0)

Create a file begin control message.

Parameters:

Name Type Description Default
path str

Relative path.

 required
size int

Total file size.

 required
compression str

Compression type. Default is 'none'.

 'none'
append_from int

Byte offset to append from. Default is 0.

 0

Returns:

Type Description
str

JSON string of the message.
Source code in src/p2p_copy/protocol.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
def file_begin(path: str, size: int, compression: str = "none", append_from: int = 0) -> str:
    """
    Create a file begin control message.

    Parameters
    ----------
    path : str
        Relative path.
    size : int
        Total file size.
    compression : str, optional
        Compression type. Default is 'none'.
    append_from : int, optional
        Byte offset to append from. Default is 0.

    Returns
    -------
    str
        JSON string of the message.
    """
    """
    Start of a file stream. If append_from is given, it indicates the sender will
    only send bytes from [append_from .. size) and the receiver should open in 'ab'.
    """
    msg: Dict[str, Any] = {
        "type": "file",
        "path": path,
        "size": int(size),
        "compression": compression,
        "append_from": append_from
    }

    return dumps(msg)

encrypted_file_begin 

encrypted_file_begin(hidden_file_info)

Wrap encrypted file info in a control message.

Parameters:

Name Type Description Default
hidden_file_info bytes

Encrypted file begin data.

 required

Returns:

Type Description
str

JSON string of the enc_file message.
Source code in src/p2p_copy/protocol.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def encrypted_file_begin(hidden_file_info: bytes) -> str:
    """
    Wrap encrypted file info in a control message.

    Parameters
    ----------
    hidden_file_info : bytes
        Encrypted file begin data.

    Returns
    -------
    str
        JSON string of the enc_file message.
    """
    payload = {
        "type": "enc_file",
        "hidden_file": hidden_file_info.hex()
    }
    return dumps(payload)

pack_chunk 

pack_chunk(seq, chain, payload)

Pack a chunk into a binary frame.

Parameters:

Name Type Description Default
seq int

Sequence number.

 required
chain bytes

32-byte chain checksum.

 required
payload bytes

The data payload.

 required

Returns:

Type Description
bytes

Packed frame.
Source code in src/p2p_copy/protocol.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def pack_chunk(seq: int, chain: bytes, payload: bytes) -> bytes:
    """
    Pack a chunk into a binary frame.

    Parameters
    ----------
    seq : int
        Sequence number.
    chain : bytes
        32-byte chain checksum.
    payload : bytes
        The data payload.

    Returns
    -------
    bytes
        Packed frame.
    """
    return CHUNK_HEADER.pack(seq, chain) + payload

unpack_chunk 

unpack_chunk(frame)

Unpack a binary chunk frame.

Parameters:

Name Type Description Default
frame bytes

The binary frame.

 required

Returns:

Type Description
Tuple[int, bytes, bytes]

(seq, chain, payload)

Raises:

Type Description
ValueError

If frame is too short.
Source code in src/p2p_copy/protocol.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def unpack_chunk(frame: bytes) -> Tuple[int, bytes, bytes]:
    """
    Unpack a binary chunk frame.

    Parameters
    ----------
    frame : bytes
        The binary frame.

    Returns
    -------
    Tuple[int, bytes, bytes]
        (seq, chain, payload)

    Raises
    ------
    ValueError
        If frame is too short.
    """
    if len(frame) < CHUNK_HEADER.size:
        raise ValueError("short chunk frame")
    seq, chain = CHUNK_HEADER.unpack(frame[:CHUNK_HEADER.size])
    payload = frame[CHUNK_HEADER.size:]
    return seq, chain, payload

 

p2p_copy.io_utils

read_in_chunks async 

read_in_chunks(fp, *, chunk_size=CHUNK_SIZE)

Asynchronously read bytes from a file in chunks.

Parameters:

Name Type Description Default
fp BinaryIO

The file pointer to read from.

 required
chunk_size int

Size of each chunk in bytes. Default is 1 MiB.

 CHUNK_SIZE

Yields:

Type Description
bytes

The next chunk of data.
Source code in src/p2p_copy/io_utils.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
async def read_in_chunks(fp: BinaryIO, *, chunk_size: int = CHUNK_SIZE) -> AsyncIterable[bytes]:
    """
    Asynchronously read bytes from a file in chunks.

    Parameters
    ----------
    fp : BinaryIO
        The file pointer to read from.
    chunk_size : int, optional
        Size of each chunk in bytes. Default is 1 MiB.

    Yields
    ------
    bytes
        The next chunk of data.
    """

    while True:
        # Read from disk without blocking the event-loop
        chunk = await asyncio.to_thread(fp.read, chunk_size)
        if not chunk:
            break
        yield chunk

compute_chain_up_to async 

compute_chain_up_to(path, limit=None)

Compute chained checksum over the raw bytes of a file up to a limit.

Parameters:

Name Type Description Default
path Path

Path to the file.

 required
limit int

Maximum bytes to hash. If None, hash the entire file.

 None

Returns:

Type Description
tuple[int, bytes]

(bytes_hashed, final_chain_bytes)
Source code in src/p2p_copy/io_utils.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def compute_chain_up_to(path: Path, limit: int | None = None) -> Tuple[int, bytes]:
    """
    Compute chained checksum over the raw bytes of a file up to a limit.

    Parameters
    ----------
    path : Path
        Path to the file.
    limit : int, optional
        Maximum bytes to hash. If None, hash the entire file.

    Returns
    -------
    tuple[int, bytes]
        (bytes_hashed, final_chain_bytes)
    """

    c = ChainedChecksum()
    hashed = 0
    with path.open("rb") as fp:
        if limit is None:
            while True:
                chunk = await asyncio.to_thread(fp.read, CHUNK_SIZE)
                if not chunk:
                    break
                hashed += len(chunk)
                c.next_hash(chunk)
        else:
            remaining = int(limit)
            while remaining > 0:
                to_read = min(remaining, CHUNK_SIZE)
                chunk = await asyncio.to_thread(fp.read, to_read)
                if not chunk:
                    break
                hashed += len(chunk)
                remaining -= len(chunk)
                c.next_hash(chunk)
    return hashed, c.prev_chain

iter_manifest_entries 

iter_manifest_entries(paths, n=1)

Collect manifest entries for files in the given paths (files or directories) and partition into n groups with approximately equal total sizes.

Parameters:

Name Type Description Default
paths List[str]

List of file or directory paths.

 required
n int

Number of groups to partition into (default 1).

 1

Returns:

Type Description
List[List[Tuple[Path, Path, int]]]

n lists, each containing (absolute_path, relative_path, size) tuples with roughly equal total size. If n=1, returns [[all_entries]].
Notes
  • Entries are sorted by size descending for partitioning.
  • Skips non-existent or invalid paths.
  • Uses greedy heuristic for partitioning.
Source code in src/p2p_copy/io_utils.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def iter_manifest_entries(paths: List[str], n: int = 1) -> List[List[Tuple[Path, Path, int]]]:
    """
    Collect manifest entries for files in the given paths (files or directories) and partition into n groups
    with approximately equal total sizes.

    Parameters
    ----------
    paths : List[str]
        List of file or directory paths.
    n : int, optional
        Number of groups to partition into (default 1).

    Returns
    -------
    List[List[Tuple[Path, Path, int]]]
        n lists, each containing (absolute_path, relative_path, size) tuples with roughly equal total size.
        If n=1, returns [[all_entries]].

    Notes
    -----
    - Entries are sorted by size descending for partitioning.
    - Skips non-existent or invalid paths.
    - Uses greedy heuristic for partitioning.
    """

    if not isinstance(paths, list):
        print("[p2p_copy] send(): files or dirs must be passed as list")
        return [[] for _ in range(n)]
    elif not paths:
        return [[] for _ in range(n)]
    if n <= 0:
        raise ValueError("n must be positive")

    # Collect all entries
    entries = []
    for raw in paths:
        if len(raw) == 1:
            print("[p2p_copy] send(): probably not a file:", raw)
            continue
        p = Path(raw).expanduser()
        if not p.exists():
            print("[p2p_copy] send(): file does not exist:", p)
            continue
        if p.is_file():
            entries.append((p.resolve(), Path(p.name), p.stat().st_size))
        else:
            root = p.resolve()
            for sub in sorted(root.rglob("*")):
                if sub.is_file():
                    rel = Path(p.name) / sub.relative_to(root)
                    entries.append((sub.resolve(), rel, sub.stat().st_size))

    # Deduplicate by absolute path (keep first occurrence)
    seen = set()
    unique_entries = []
    for entry in entries:
        abs_p = entry[0]
        if abs_p not in seen:
            seen.add(abs_p)
            unique_entries.append(entry)
    entries = unique_entries

    if not entries:
        return [[] for _ in range(n)]

    # Sort by size descending
    entries.sort(key=lambda x: x[2], reverse=True)

    # Use a min-heap to track current group sums and indices
    groups = [[] for _ in range(n)]
    heap = [(0, i) for i in range(n)]  # (current_sum, group_index)
    heapq.heapify(heap)

    for tup in entries:
        size = tup[2]
        current_sum, group_idx = heapq.heappop(heap)
        groups[group_idx].append(tup)
        new_sum = current_sum + size
        heapq.heappush(heap, (new_sum, group_idx))
    return groups

ensure_dir 

ensure_dir(p)

Ensure the directory exists, creating parents if needed.

Parameters:

Name Type Description Default
p Path

The path to ensure is a directory.

 required
Source code in src/p2p_copy/io_utils.py 
161
162
163
164
165
166
167
168
169
170
def ensure_dir(p: Path) -> None:
    """
    Ensure the directory exists, creating parents if needed.

    Parameters
    ----------
    p : Path
        The path to ensure is a directory.
    """
    p.mkdir(parents=True, exist_ok=True)

 

For api usage examples, see APIExamples. For feature details, see Features.