diff --git a/README.md b/README.md index 6ea3977..978119f 100644 --- a/README.md +++ b/README.md @@ -71,9 +71,14 @@ CryptoLayer использует проверенные технологии и * **[CryptoLayer CLI](https://github.com/igmunv/cryptolayer-cli)** — официальный интерфейс для командной строки. Отлично подойдёт для работы в терминале. -### Хотите добавить свой проект? -Если вы разработали приложение с использованием **CryptoLayer**, мы с радостью добавим его в этот список! -Просто создайте Pull Request, указав название проекта, краткое описание и ссылку на репозиторий. +* **[CryptoLayer Web UI](https://github.com/DaPon4ik/cryptolayer-webui)** — web-интерфейс для защищенного обмена сообщениями в мессенджерах. Удобный и красивый интерфейс. + +* **[zkgram](https://github.com/Gerate-Technik/zkgram)** — приватный клиент Telegram. Подойдёт если необходимо удобное и безопасное общение только в Telegram. + +> [!NOTE] +> **Хотите добавить свой проект?** +> +> Если вы разработали приложение с использованием **CryptoLayer**, мы с радостью добавим его в этот список! Просто создайте Pull Request, указав название проекта, краткое описание и ссылку на репозиторий. ## Документация diff --git a/README_en.md b/README_en.md new file mode 100644 index 0000000..3e10864 --- /dev/null +++ b/README_en.md @@ -0,0 +1,95 @@ +
+ +CryptoLayer Logo + +
+

CryptoLayer

+
A cryptographic layer that operates on top of existing messengers, providing end-to-end message encryption solely on the user's side
+ +[![License](https://img.shields.io/badge/License-MIT-brightgreen?color=orange&style=flat-square)](LICENSE) +[![Contributions Welcome](https://img.shields.io/badge/Contributions-Welcome-brightgreen?style=flat-square)](CONTRIBUTING.md) +
+[![Python](https://img.shields.io/badge/python-3670A0?style=for-the-badge&logo=python&logoColor=ffdd54&style=flat-square)](https://www.python.org/) +
+
+[Русский](README.md) • English + +
+ + +## What is CryptoLayer? + +**CryptoLayer** is a library that does not replace messengers but protects the content of your messages using cryptography. + +
+
+How CryptoLayer Works +
+
+ +Simply put: for CryptoLayer, any messenger is **just an untrusted "wire"**, so all encryption and delivery guarantees happen exclusively within CryptoLayer **only on your device**. + +
+
+How Looks Data Transfer +
+
+ +## Custom Pseudo-Network Stack + +The library implements its own pseudo-network stack: + +
+
+Net Levels +
+
+ +## Full Modularity + +The main feature of CryptoLayer is its modularity! The communication channel can be anything: + +- **Messengers**: Telegram, VK, Discord... +- **Network protocols**: HTTP, SSH, FTP, UDP... +- **Clouds and services**: Google Drive, Yandex Disk, YouTube comments, streaming platforms... +- **And other types**: clipboard, file system, Bluetooth... + +### Anything! [Just write a module!](docs/README.md#5-модули) + +The library **doesn't care** how bytes are transmitted. For it, **any** messenger, protocol, or service is simply an **untrusted "wire"**. + +## Technology and Security + +CryptoLayer uses proven technologies and methods to ensure the security of your data: + +- **Encryption** - AES-256-GCM for content protection +- **Digital signatures and integrity verification** - ECDSA (SECP256R1 curve) for data signing +- **Key exchange** - ECDH protocol (SECP256R1 curve, X9.62 compressed point format) +- **Obfuscation** - custom byte-to-word encoding (WordCoder) to bypass basic messenger filters + +## Ecosystem and Ready-made Applications + +* **[CryptoLayer CLI](https://github.com/igmunv/cryptolayer-cli)** — official command-line interface. Great for terminal use. + +* **[CryptoLayer Web UI](https://github.com/DaPon4ik/cryptolayer-webui)** — web interface for secure message exchange in messengers. Convenient and beautiful UI. + +* **[zkgram](https://github.com/Gerate-Technik/zkgram)** — private Telegram client. Suitable if you need convenient and secure communication exclusively in Telegram. + +> [!NOTE] +> **Want to add your project?** +> +> If you have developed an application using **CryptoLayer**, we will be happy to add it to this list! Simply create a Pull Request with the project name, a brief description, and a link to the repository. + +## Documentation + +In the [documentation](docs/README.md) you will find more information about CryptoLayer: + +- **How the library works** +- **How to use it** in your code +- **Architecture** of CryptoLayer + +## IMPORTANT + +The user has a fundamental right to **private and secure communication**. This includes the right to **independently use cryptographic means** to protect their messages, as well as the right to confidentiality of correspondence **without unauthorized access by third parties**. + +The project proceeds from the principle that secure and private communication is a **basic digital norm, not a privilege**. diff --git a/bench/bench.py b/bench/bench.py new file mode 100644 index 0000000..51c3f24 --- /dev/null +++ b/bench/bench.py @@ -0,0 +1,126 @@ +"""Benchmark the CryptoLayer send/receive transform pipeline. + +Measures every stage that touches a message on its way out and back in, +excluding threading/polling overhead, so the pure CPU cost is visible. + +Run: python3 bench/bench.py +""" +import os +import statistics +import sys +import time + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +import brotli +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + +import config +from levels.packet import ApplicationPacket, DataTypes, PackTypes, TextMessagePacket, TransportPacket +from wordcoder import WordCoder + +# 256 distinct <=10-char words, same shape as the real dictionary repo. +SYLL_A = ["ba", "ve", "gi", "do", "zhu", "ki", "la", "mo", "ne", "pu", "ra", "so", "tu", "fi", "ha", "che"] +SYLL_B = ["lom", "ves", "gor", "dym", "zhar", "kit", "lug", "mox", "nos", "puh", "rov", "sud", "tir", "fon", "hor", "chan"] +WORDCODER_DICT = { + f"{a * 16 + b:02x}": SYLL_A[a] + SYLL_B[b] + for a in range(16) + for b in range(16) +} + +PAYLOADS = { + "short (32B)": b"x" * 32, + "chat (256B)": ("Wake me up when september ends. " * 8).encode(), + "long (4KB)": (os.urandom(16).hex() * 256).encode()[:4096], +} + + +def timeit(fn, *, min_rounds=50, min_seconds=0.25): + """Return (median_seconds, rounds). Warms up, then loops until stable.""" + for _ in range(3): + fn() + samples = [] + deadline = time.perf_counter() + min_seconds + while len(samples) < min_rounds or time.perf_counter() < deadline: + t0 = time.perf_counter() + fn() + samples.append(time.perf_counter() - t0) + return statistics.median(samples), len(samples) + + +def fmt(seconds): + if seconds >= 1: + return f"{seconds:8.3f} s " + if seconds >= 1e-3: + return f"{seconds * 1e3:8.3f} ms" + return f"{seconds * 1e6:8.3f} us" + + +def main(): + wc = WordCoder(WORDCODER_DICT) + aes_key = os.urandom(32) + aesgcm = AESGCM(aes_key) + sign_key = ec.generate_private_key(ec.SECP256R1()) + verify_key = sign_key.public_key() + chunk_size = config.CHUNK_SIZE + + print(f"CHUNK_SIZE={chunk_size} COMPRESS_QUALITY={config.COMPRESS_QUALITY}") + + for name, raw in PAYLOADS.items(): + # --- stage inputs, mirroring the real pipeline --- + app_bytes = ApplicationPacket( + PackTypes.COMMUNIC.value, + DataTypes.TEXT.value, + TextMessagePacket(int(time.time()), raw).to_bytes(), + ).to_bytes() + + compressed = brotli.compress(app_bytes, quality=config.COMPRESS_QUALITY) + nonce = os.urandom(12) + encrypted = nonce + aesgcm.encrypt(nonce, compressed, associated_data=None) + chunks = [encrypted[i:i + chunk_size] for i in range(0, len(encrypted), chunk_size)] + n_chunks = len(chunks) + transport_packets = [ + TransportPacket(0x0, 0, n_chunks, i, int(time.time()), c).to_bytes() + for i, c in enumerate(chunks) + ] + signature = sign_key.sign(transport_packets[0], ec.ECDSA(hashes.SHA256())) + signed = len(signature).to_bytes(1, "big") + signature + transport_packets[0] + wire = " ".join(wc.encode(signed)) + + print(f"\n=== {name} ===") + print(f" app packet {len(app_bytes)}B -> brotli {len(compressed)}B " + f"-> +aesgcm {len(encrypted)}B -> {n_chunks} chunk(s) " + f"-> wire {len(wire)} chars ({len(wire) / len(raw):.1f}x expansion)") + + stages = [ + # (label, callable, how many times it runs per message) + ("brotli.compress", lambda: brotli.compress(app_bytes, quality=config.COMPRESS_QUALITY), 1), + ("brotli.decompress", lambda: brotli.decompress(compressed), 1), + ("aesgcm.encrypt", lambda: aesgcm.encrypt(nonce, compressed, associated_data=None), 1), + ("aesgcm.decrypt", lambda: aesgcm.decrypt(nonce, encrypted[12:], associated_data=None), 1), + ("ecdsa.sign", lambda: sign_key.sign(transport_packets[0], ec.ECDSA(hashes.SHA256())), n_chunks), + ("ecdsa.verify", lambda: verify_key.verify(signature, transport_packets[0], ec.ECDSA(hashes.SHA256())), n_chunks), + ("wordcoder.encode", lambda: " ".join(wc.encode(signed)), n_chunks), + ("wordcoder.decode", lambda: wc.decode(wire.split(" ")), n_chunks), + ] + + total_send = 0.0 + total_recv = 0.0 + print(f" {'stage':<20} {'median':>12} {'xN':>4} {'per message':>13}") + for label, fn, times in stages: + median, _ = timeit(fn) + per_msg = median * times + if "compress" in label and "de" not in label or label in ("aesgcm.encrypt", "ecdsa.sign", "wordcoder.encode"): + total_send += per_msg + else: + total_recv += per_msg + print(f" {label:<20} {fmt(median)} x{times:<3} {fmt(per_msg)}") + print(f" {'TOTAL send':<20} {'':>12} {fmt(total_send)}") + print(f" {'TOTAL recv':<20} {'':>12} {fmt(total_recv)}") + print(f" {'THROUGHPUT send':<20} {'':>12} {len(raw) / total_send / 1024:8.1f} KiB/s") + + +if __name__ == "__main__": + main() diff --git a/bench/e2e.py b/bench/e2e.py new file mode 100644 index 0000000..29d358c --- /dev/null +++ b/bench/e2e.py @@ -0,0 +1,175 @@ +"""End-to-end CryptoLayer benchmark: two real peers over an in-memory channel. + +Measures handshake time, per-message latency and sustained throughput +through the full stack (brotli + AES-GCM + chunking + ACK + ECDSA + WordCoder). + +Run: python3 bench/e2e.py [--latency MS] [--messages N] [--size BYTES] +""" +import argparse +import logging +import os +import shutil +import statistics +import sys +import tempfile +import threading +import time + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) +sys.path.insert(0, os.path.dirname(__file__)) + +from loopback import Loopback # noqa: E402 (needs sys.path above) + +import config # noqa: E402 +from UIProvider import UIProvider # noqa: E402 + +SYLL_A = ["ba", "ve", "gi", "do", "zhu", "ki", "la", "mo", "ne", "pu", "ra", "so", "tu", "fi", "ha", "che"] +SYLL_B = ["lom", "ves", "gor", "dym", "zhar", "kit", "lug", "mox", "nos", "puh", "rov", "sud", "tir", "fon", "hor", "chan"] +WORDCODER_DICT = { + f"{a * 16 + b:02x}": SYLL_A[a] + SYLL_B[b] + for a in range(16) + for b in range(16) +} + +TEXT = ("Slushay, ya zapushil fiks v main, CI zelyonyy. Proverь pozhaluysta " + "handshake na svoey storone, u menya ne vosproizvoditsya nikak. ") + + +class SilentUI(UIProvider): + """Auto-trusting UI so the handshake runs unattended.""" + + def __init__(self, name): + self.name = name + self.ready = threading.Event() + self.received = [] + self.recv_lock = threading.Lock() + + def request_data(self, prompt, data_type): + return "" + + def update_status(self, stage, message, status_type="in_progress"): + pass + + def on_text_received(self, timestamp, text): + with self.recv_lock: + self.received.append((time.perf_counter(), text)) + + def check_signatures(self, my_sign, companion_sign): + return True + + def on_ready(self): + self.ready.set() + + def on_ping_timeout(self): + pass + + def on_disconnect(self): + pass + + +def build_pair(latency_s, data_root): + """Return (peer_a, peer_b, ui_a, ui_b) with the handshake already done.""" + from crypto_layer import CryptoLayer + + mod_a = Loopback(latency_s, "A") + mod_b = Loopback(latency_s, "B") + mod_a.peer_inbox = mod_b.inbox + mod_b.peer_inbox = mod_a.inbox + + ui_a, ui_b = SilentUI("A"), SilentUI("B") + peer_a = CryptoLayer(ui_a, os.path.join(data_root, "a"), mod_a, "hunter2", WORDCODER_DICT) + peer_b = CryptoLayer(ui_b, os.path.join(data_root, "b"), mod_b, "hunter2", WORDCODER_DICT) + + errors = [] + + def run(peer): + try: + peer.init() + except Exception as exc: # surfaced by the caller, never swallowed + errors.append(exc) + + threads = [threading.Thread(target=run, args=(p,), daemon=True) for p in (peer_a, peer_b)] + t0 = time.perf_counter() + for t in threads: + t.start() + for t in threads: + t.join(timeout=120) + handshake = time.perf_counter() - t0 + + if errors: + raise errors[0] + if not (ui_a.ready.is_set() and ui_b.ready.is_set()): + raise RuntimeError("handshake did not complete within 120s") + + return peer_a, peer_b, ui_a, ui_b, handshake, (mod_a, mod_b) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--latency", type=float, default=0.0, help="one-way channel latency, ms") + ap.add_argument("--messages", type=int, default=20) + ap.add_argument("--size", type=int, default=256, help="plaintext bytes per message") + ap.add_argument("--incompressible", action="store_true", + help="use high-entropy payloads (hashes, links, base64) so brotli cannot\ncollapse them into a single chunk") + args = ap.parse_args() + + logging.disable(logging.CRITICAL) + latency_s = args.latency / 1000.0 + if args.incompressible: + import random + payload = "".join(random.Random(7).choices("0123456789abcdef", k=args.size)) + else: + payload = (TEXT * ((args.size // len(TEXT)) + 1))[:args.size] + + data_root = tempfile.mkdtemp(prefix="cl-bench-") + try: + print(f"config: CHUNK_SIZE={config.CHUNK_SIZE} (transport hardcodes its own) " + f"COMPRESS_QUALITY={config.COMPRESS_QUALITY}") + print(f"run: {args.messages} msg x {args.size}B, one-way latency {args.latency}ms\n") + + peer_a, peer_b, ui_a, ui_b, handshake, mods = build_pair(latency_s, data_root) + print(f"handshake: {handshake:7.3f} s") + + chunk_size = peer_a.TRANSPORT_LEVEL.CHUNK_SIZE + print(f"transport chunk: {chunk_size} B") + + base_sent = sum(m.sent_messages for m in mods) + base_chars = sum(m.sent_chars for m in mods) + + latencies = [] + t_start = time.perf_counter() + for i in range(args.messages): + sent_at = time.perf_counter() + want = i + 1 + peer_a.send(payload) + while len(ui_b.received) < want: + if time.perf_counter() - sent_at > 120: + raise RuntimeError(f"message {i} never arrived") + time.sleep(0.001) + latencies.append(ui_b.received[want - 1][0] - sent_at) + wall = time.perf_counter() - t_start + + assert all(text == payload for _, text in ui_b.received), "payload corrupted in transit" + + wire_msgs = sum(m.sent_messages for m in mods) - base_sent + wire_chars = sum(m.sent_chars for m in mods) - base_chars + + print(f"\nlatency per message ({args.messages} samples)") + print(f" min {min(latencies) * 1e3:7.1f} ms") + print(f" median {statistics.median(latencies) * 1e3:7.1f} ms") + print(f" max {max(latencies) * 1e3:7.1f} ms") + print(f"\nthroughput") + print(f" wall {wall:7.3f} s") + print(f" messages/s {args.messages / wall:7.2f}") + print(f" goodput {args.messages * args.size / wall / 1024:7.2f} KiB/s") + print(f"\nchannel cost (both directions, incl. ACKs)") + print(f" channel msgs {wire_msgs:7d} ({wire_msgs / args.messages:.1f} per user message)") + print(f" channel chars {wire_chars:7d} ({wire_chars / (args.messages * args.size):.1f}x plaintext)") + + peer_a.stop(send_disconnect=False) + finally: + shutil.rmtree(data_root, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/bench/loopback.py b/bench/loopback.py new file mode 100644 index 0000000..a9a1564 --- /dev/null +++ b/bench/loopback.py @@ -0,0 +1,77 @@ +"""In-memory BaseModule that wires two CryptoLayer peers together. + +Stands in for a real messenger module so the whole stack can be exercised +without a network. One-way latency is configurable to model a real channel. +""" +import queue +import random +import threading +import time + +try: + from base_module import BaseModule, Credential +except ImportError as exc: # pragma: no cover + raise SystemExit( + "base_module is missing. It is declared in pyproject.toml; install the " + "project dependencies first, e.g. `pip install -e .`" + ) from exc + + +class Loopback(BaseModule): + """A module whose channel is a queue owned by its peer. + + Set `peer_inbox` after construction; whatever `send` writes lands there + and the peer's listener thread feeds it into its transitional level. + """ + + name = "Loopback" + description = "In-memory channel for benchmarking" + unique_id = "bench.loopback_0001" + expected_credentials = [Credential("None", "unused")] + + def __init__(self, latency_s=0.0, label="?", loss=0.0, seed=0, record=False): + super().__init__() + self.inbox = queue.Queue() + self.peer_inbox = None + self.latency_s = latency_s + self.label = label + self.loss = loss + self.record = record + self.sent_messages = 0 + self.sent_chars = 0 + self.dropped = 0 + self.wire = [] + self._rng = random.Random(seed) + self._rng_lock = threading.Lock() + + def create_session(self, ingester): + module = self + + class Sender(BaseModule.Sender): + def send(self, text: str): + module.sent_messages += 1 + module.sent_chars += len(text) + if module.record: + module.wire.append(text) + if module.loss: + with module._rng_lock: + drop = module._rng.random() < module.loss + if drop: + module.dropped += 1 + return + if module.latency_s: + time.sleep(module.latency_s) + module.peer_inbox.put(text) + + class Listener(BaseModule.Listener): + def listen(self): + while not self.stop_event.is_set(): + try: + text = module.inbox.get(timeout=0.05) + except queue.Empty: + continue + self.ingester(text) + + self.sender = Sender([], None) + self.listener = Listener([], ingester, None, self.stop_event) + threading.Thread(target=self.listener.listen, daemon=True).start() diff --git a/docs/README.md b/docs/README.md index 1f9f957..ff7e5e9 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,5 +1,7 @@ # Документация +Русский • [English](README_en.md) + ## Содержание - [1. О CryptoLayer](#1-о-cryptolayer) diff --git a/docs/README_en.md b/docs/README_en.md new file mode 100644 index 0000000..604b37f --- /dev/null +++ b/docs/README_en.md @@ -0,0 +1,648 @@ +# Documentation + +[Русский](README.md) • English + +## Table of Contents + +- [1. About CryptoLayer](#1-about-cryptolayer) + - [1.1. What is it](#11-what-is-it) + - [1.2. Why is it needed](#12-why-is-it-needed) + - [1.3. Architecture](#13-architecture) +- [2. Usage](#2-usage) + - [2.1. Integration Guide](#21-integration-guide) + - [2.2. Basic Operations](#22-basic-operations) + - [2.3. UIProvider](#23-uiprovider) +- [3. How it works](#3-how-it-works) + - [3.1. Brief Steps](#31-brief-steps) + - [3.2. Initialization](#32-initialization) + - [3.3. Main Workflow](#33-main-workflow) + - [3.4. Shutdown](#34-shutdown) + - [3.5. Delivery Guarantee](#35-delivery-guarantee) + - [3.6. Connection Stability](#36-connection-stability) + - [3.7. Packets](#37-packets) +- [4. Data Protection](#4-data-protection) + - [4.1. Encryption of sent data](#41-encryption-of-sent-data) + - [4.2. Encryption of CryptoLayer files](#42-encryption-of-cryptolayer-files) + - [4.3. Digital signatures and data integrity](#43-digital-signatures-and-data-integrity) + - [4.4. Key exchange](#44-key-exchange) + - [4.5. Masking](#45-masking) +- [5. Modules](#5-modules) + - [5.1. Where are existing modules located](#51-where-are-existing-modules-located) + - [5.2. Creating your own module](#52-creating-your-own-module) + - [5.3. Testing your own module](#53-testing-your-own-module) + +## 1. About CryptoLayer + +### 1.1. What is it + +CryptoLayer is a library that allows you to securely exchange messages through any messenger (and not just messengers). + +The library is an independent layer between the user and the messenger, which protects transmitted data using cryptographic means. + +
+
+How It Works +
+
+ +### 1.2. Why is it needed + +In today's world, it's difficult to 100% trust existing messengers. There is no guarantee that your data will not be passed on to third parties or used by the messenger owners. + +You might immediately think about creating your own mega-secure messenger, but it could simply be blocked or forced to hand over user messages. That's why CryptoLayer **uses existing messengers**, and it uses them **simply as a communication line** (a wire) which is not trusted. + +
+
+How Looks Data Transfer +
+
+ +### 1.3. Architecture + +The library consists of three main parts: **Manager**, **pseudo-network layers**, and **modules**. + +
+
+Architecture +
+
+ +#### Manager + +Manages all the logic, as well as the pseudo-network layers. Performs initial setup and initialization of the library. Responsible for communication with the UI. + +#### Pseudo-network layers + +
+
+Net Levels +
+
+ +Analogous to the TCP/IP model. Each layer performs its specific function and then passes the data to the next layer: + +- **Application Layer** - provides convenient functions to the manager, such as sending a text message. Packages everything into an application packet with necessary fields for convenient data handling at this layer. +- **Presentation Layer** - compresses data for more efficient transmission. Also encrypts data before sending and decrypts it after receiving. +- **Transport Layer** - provides guaranteed data delivery by splitting it into chunks, packaging a chunk into a transport packet, and then sending the packet while waiting for an acknowledgment of receipt. +- **Transition Layer** - signs outgoing data and verifies the signature of incoming data. Also, before sending to the module, it encodes data bytes into words, based on the principle 1 byte = 1 word - to mask the transmission of bytes. + +#### Modules + +Implement the interface for interacting with a specific data transmission channel - messenger, service, or protocol. + +This is where sending and receiving data happens, by accessing the API of a specific messenger or other communication channel. + +You can use anything as a data transmission channel: network protocols (http, ssh, ftp, tcp, udp), services (cloud drives, streaming platforms), messengers, file systems - **anything!** The main thing is to write a module. + +Modularity allows using CryptoLayer with any messengers, protocols, and services. The main thing is that a module exists for the specific data transmission channel. If it doesn't exist, it can always be developed. + +## 2. Usage + +### 2.1. Integration Guide + +#### 1. Add CryptoLayer to the project: + +Add the library to the project as a Git submodule: + +```bash +git submodule add https://github.com/igmunv/cryptolayer cryptolayer + +git add .gitmodules cryptolayer/ + +git commit -m "Add new submodule: cryptolayer" +``` + +OR, if you don't want to deal with Git, download the latest version of the library: + +https://github.com/igmunv/cryptolayer/releases/latest + +and then unpack it into the project directory. + +#### 2. Add CryptoLayer modules to the project + +Do pretty much the same as in the previous step. Add the CryptoLayer modules repository to the project as a Git submodule: + +```bash +git submodule add https://github.com/igmunv/cryptolayer-modules modules + +git add .gitmodules modules/ + +git commit -m "Add new submodule: cryptolayer-modules" +``` + +OR, if you don't want to deal with Git, download the repository. + +If you will use a different collection of modules, simply replace the URL. + +#### 3. Add the library to the project configuration files: + +At the end of the `requirements.txt` file, or if using `pyproject.toml`, in the `dependencies` list field, add the following dependencies: + +``` +-e ./cryptolayer +cryptolayer-module-interface @ git+https://github.com/igmunv/cryptolayer-module-interface.git +``` + +Here we added CryptoLayer as a Python library, as well as `cryptolayer-module-interface` for working with modules. + +#### 4. Import libraries in code: + +Import the libraries added in the previous step into your code: + +```python +from crypto_layer import CryptoLayer +from UIProvider import UIProvider +from base_module import BaseModule +``` + +Also, it's worth importing the modules right away (it's normal that the hidden_imports.py file doesn't exist yet): + +```python +import modules.hidden_imports +``` + +#### 5. Implementing UIProvider: + +Before creating an instance of the `CryptoLayer` class, you need to implement the `UIProvider` class, which acts as an intermediary between CryptoLayer and your application with UI. + +The `UIProvider` class is located in the `UIProvider.py` file in the CryptoLayer directory. + +#### 6. Module: + +Before creating an instance of the `CryptoLayer` class, you need to select a module, which is then passed as an argument when creating the `CryptoLayer` class. + +You need to implement a module selection by the user, or statically use one specific module. + +Discovering all modules can be done as follows: + +```python +# Path to the submodule directory with modules +MODULES_DIR_PATH = "modules" + +# Iterate over all elements in the directory +for item in os.listdir(modules_path): + + # Get the full correct path of the element + item_path = os.path.join(MODULES_DIR_PATH, item) + + # Check if it's a directory (since the module is in a directory) + # And that the path doesn't start with '_' (to exclude certain directories) + if os.path.isdir(item_path) and not item.startswith('_'): + + try: + + # Try to import the module + module = importlib.import_module(f"{item}.main") + + # Iterate over all objects in the imported module + for name, obj in inspect.getmembers(module, inspect.isclass): + + # Look for an object inherited from BaseModule, but excluding BaseModule itself + if issubclass(obj, BaseModule) and obj is not BaseModule: + + # Get the module class + module_class = obj() + + # We can add the module to the common list of modules + MODULES.append(module_class) +``` + +After this, the modules in the `MODULES` variable can be used, for example, for the user to select a specific module. + +Each module has `name` and `description` fields, which can be listed for the user to choose a specific module. Modules also have a `unique_id` field, which is a unique module identifier. This field can be used, for example, to save some information about a specific module to a file. + +#### 7. Byte-Word dictionary for WordCoder: + +You need to prepare a dictionary for encoding bytes into words. It's best to give the user the ability to choose dictionaries and create their own, since different users may use different programs for working with CryptoLayer, and these programs may have their own custom dictionaries. + +Ready-made dictionaries are available in this repository (you can also add it to the project as a submodule and then select the desired one): + +https://github.com/igmunv/cryptolayer-wordcoder-dicts + +#### 8. Creating an instance of the CryptoLayer class: + +Before creating, you need to prepare the following variables: + +- `ui_provider` - the implemented UIProvider class. Needed for communication between the application and CryptoLayer. +- `data_dir` - the path to the data storage directory. Needed for CryptoLayer to save its data there. +- `module_class` - the module class. CryptoLayer will use it as the module. +- `password` - the password. Used when CryptoLayer saves data to a file, to encrypt the contents (if the user forgets the password, the directory at `data_dir` needs to be deleted). +- `wordcoder_dict` - the byte-word dictionary. Needed for the WordCoder component to encode bytes into words for masking. + +When all variables are ready, you can create an object of the CryptoLayer class: + +```python +clayer = CryptoLayer(ui_provider, data_dir, module_class, password, wordcoder_dict) +``` + +After creation, you need to start the CryptoLayer initialization: + +```python +clayer.init() +``` + +And wait for CryptoLayer to be ready. When ready, the `on_ready` function in the `UIProvider` class will be called. + +#### 9. Running the project: + +It is necessary to run in the following order (it's convenient to combine all commands into one file, for example `run.sh`): + +- Update submodules: + +```bash +git submodule update --init --recursive +``` + +- Run the script to generate the modules' dependencies file: + +```bash +python3 modules/generate_reqs.py +``` + +- Run the script to generate the modules' import dependencies file (especially necessary when building the project into a binary file using PyInstaller): + +```bash +python3 modules/generate_hidden_imports.py +``` + +- Install module dependencies: + +```bash +pip install -r modules/common_requirements.txt +``` + +- Install project dependencies: + +```bash +pip install -r requirements.txt +``` + +or, if using `pyproject.toml`: + +```bash +pip install . +``` + +- Run the project: + +```bash +python3 main.py # or your entry point +``` + +### 2.2. Basic Operations + +#### Sending a message: + +To send a message, you need to call the `send` method and pass the string as an argument: + +```python +clayer.send(user_message) +``` + +#### Receiving a message: + +When a message is received, CryptoLayer will call the `on_text_received` function of your UIProvider and pass the message sending time, in Unix Time Stamp format, as well as the text message itself: + +```python +class UIProvider: +... +def on_text_received(self, timestamp: int, text: str): +... +``` + +#### Ending a session / Exiting the program: + +Before ending the communication session with the current interlocutor or before exiting the program, you need to stop the current instance of the CryptoLayer class using the `stop` function: + +```python +clayer.stop() +``` + +The function will send a `DISCONNECT` packet to the interlocutor, indicating that we are exiting and ending communication, and will also stop all threads and pseudo-network layers. + +If you don't want a `DISCONNECT` packet to be sent, pass `False` to the `send_disconnect` argument (this may be needed if the interlocutor is unavailable, i.e., `on_ping_timeout` was called): + +```python +clayer.stop(send_disconnect=False) +``` + +#### Interlocutor unavailable (Ping timeout): + +If CryptoLayer detects that the interlocutor is unavailable, it will call the `on_ping_timeout` function of your UIProvider: + +```python +class UIProvider: +... +def on_ping_timeout(self): +... +``` + +### 2.3. UIProvider + +The `UIProvider` class is needed for CryptoLayer to pass data to the application and its UI. This includes, for example, the current status, a new message from the interlocutor, an error signal, etc. + +When creating an application based on CryptoLayer, you must implement `UIProvider`. + +Let's look at each function in UIProvider that needs to be implemented: + +#### `def request_data(self, prompt: str, data_type: type)`: + +This function is intended for requesting data from the user. The `prompt` argument contains the text that should be displayed when requesting data, and `data_type` is the type of data the function should return. You must return data with the type specified in `data_type`. + +#### `def update_status(self, stage: str, message: str, status_type: str = "in_progress")`: + +This function updates the current loading and working status of CryptoLayer. `stage` contains the stage CryptoLayer is at, `message` contains more detailed information about the operation being performed, and `status_type` is the type of status. There are three status types: `in_progress`, `success`, `error`. For example, depending on the type, you could display the status in different colors. No return value is required. + +#### `def on_text_received(self, timestamp: int, text: str)`: + +This function is called when a new message arrives from the interlocutor. `timestamp` is the message sending time, in Unix Time Stamp format, and `text` is the text message itself. No return value is required. + +#### `def check_signatures(self, my_sign: str, companion_sign: str) -> bool`: + +Called for the user to verify that the signatures are correct. `my_sign` contains the user's signature, and `companion_sign` contains the interlocutor's signature. You need to ask the user if the interlocutor's signature is correct. You must return `True` if correct, `False` if not. + +#### `def on_ready(self)`: + +Called when CryptoLayer has finished initialization and is ready to exchange messages. No return value is required. + +#### `def on_ping_timeout(self)`: + +Called when the interlocutor is unavailable. CryptoLayer continues to work, and further actions depend on the application. No return value is required. + +#### `def on_disconnect(self)`: + +Called when the interlocutor disconnects and ends communication. CryptoLayer continues to work, and further actions depend on the application. No return value is required. + +## 3. How it works + +### 3.1. Brief Steps + +- Initialization + - Exchange node identifiers + - Exchange signatures + - Exchange public keys + - Generate shared key +- Main Workflow + - Sending user message + - Message goes down through the pseudo-network stack + - Message is sent to the module, and then to the communication channel + - Receiving interlocutor message + - Message arrives from the communication channel and goes to the module + - Message goes up through the pseudo-network stack + - Ping + - If there were no messages from the interlocutor for 30 seconds, a PING is sent + - If there is no response to PING within 30 seconds, the `on_ping_timeout` function is called +- Shutdown + - Send DISCONNECT packet + - Disable pseudo-network stack + +### 3.2. Initialization + +
+
+Initialization +
+
+ +After calling the `init` function, the initialization process starts (both nodes must start the initialization process): + +#### Exchange of Node IDs + +An attempt is made to read the current Node ID from the `node_id` file. If the file does not exist, the identifier is generated and written to the file. Then the identifiers are exchanged. This is needed for the next step. + +#### Exchange of Digital Signatures + +Next, work with digital signatures begins. An attempt is made to read the current signature private key from the `sign_private` file. If the file does not exist or the data could not be decrypted, a new digital signature for this node is generated. + +Then the signatures are exchanged. After that, it is checked whether the received signature of the interlocutor is known: + +In the `known_nodes` directory, a search is performed for a file whose name equals the interlocutor's `node_id`. If the file exists, the data is correctly decrypted, and the signature is known, we proceed to the encryption key exchange. If the file does not exist, the data from the file could not be decrypted, or the signature in the file does not equal the current signature of the interlocutor, then it means we have not encountered this signature before, and it needs to be verified to ensure we are indeed communicating with our interlocutor. + +The user should be shown their signature and the signature of the person they are trying to start communication with. Then both interlocutors must verify their keys through another communication channel (in-person meeting, phone call, some messenger). If the keys match, then it is definitely the person we want to communicate with. After successful verification, the signature will be written to a file and CryptoLayer will remember it. This means that verification will no longer be needed, unless the interlocutor changes their signature or Node ID. + +**This stage is the most critical!** Especially the signature verification - it cannot be ignored or skipped! You need to approach the correctness of signature verification with utmost seriousness! + +After successful signature exchange, mandatory signature verification is activated for all incoming packets. Packets with invalid signatures will be discarded and ignored. Now you can be sure that packets are sent by your interlocutor, the person you want to communicate with. **The digital signature solves the MITM problem**, and now you can safely transmit public keys! + +#### Exchange of Encryption Keys + +On both nodes, an ECC public key is generated. Then the nodes exchange keys (remember, now all packets are signed, and MITM cannot be performed) and calculate a shared secret, which is used to encrypt data using AES. + +Initialization completed successfully! Now the interlocutors can exchange messages **MAXIMUM SECURELY!** + +### 3.3. Main Workflow + +When sending, the message passes through this pipeline. I see no point in describing it in detail; everything is perfectly clear in the image: + +
+
+ACK Receive Packet +
+
+ +Also during operation, CryptoLayer checks the availability of the interlocutor: if no packets have been received from the interlocutor for 30 seconds, a Ping packet is sent at the transport layer. If the interlocutor does not respond within 30 seconds, the `on_ping_timeout` function is called, after which CryptoLayer continues to work, and further actions are left to the application using CryptoLayer. + +### 3.4. Shutdown + +When finishing work with the current instance of the CryptoLayer class, call the `stop` function. After the call, a `DISCONNECT` packet is sent to the interlocutor, indicating that we are disconnecting and ending the conversation (the current session). Then CryptoLayer waits for ALL messages in the queue to be sent, and after that, all pseudo-network layers are disconnected. + +### 3.5. Delivery Guarantee + +CryptoLayer has a packet delivery guarantee mechanism (analogous to TCP). This mechanism is implemented at the transport layer of the pseudo-network stack. + +
+
+ACK Receive Packet +
+
+ +After sending a packet, the sender does not send anything else until it receives an acknowledgment of receipt for the sent packet. After receiving the packet, the receiver calculates its hash and sends a special acknowledgment packet containing the hash of the received packet. The sender receives this acknowledgment packet, verifies the hash, and if everything is correct, proceeds to send the next packet. If the hash is incorrect, the packet is sent again. Or if after 30 seconds the receiver has not sent an acknowledgment packet, the sender sends the packet again and the waiting cycle resumes. + +### 3.6. Connection Stability + +#### Ping + +If no packets have been received from the interlocutor for 30 seconds, a PING packet (transport layer) is sent to check the interlocutor's availability. The interlocutor receives the PING packet and responds to it. If there is no response from the interlocutor within 30 seconds after sending PING, the `on_ping_timeout` function is called, but CryptoLayer continues to work. + +#### DISCONNECT + +When exiting the program or ending the current communication session, the `stop` function is called, during which a DISCONNECT packet is sent to the interlocutor, indicating the end of the current communication session. After receiving such a packet, the interlocutor's CryptoLayer will call the `on_disconnect` function of the UIProvider. In the implementation of this function, you need to finish working with the current instance of CryptoLayer. + +### 3.7. Packets + +CryptoLayer implements its own pseudo-network stack. And each pseudo-network layer has its own packet with a specific structure: + +
+
+Packet Structures +
+
+ +## 4. Data Protection + +### 4.1. Encryption of sent data + +For encrypting data sent at the presentation pseudo-network layer, the **AES-256-GCM** algorithm is used. + +### 4.2. Encryption of CryptoLayer files + +CryptoLayer saves data to a file (signatures, known interlocutors). To secure this data, encryption is used, specifically the same **AES-256-GCM** algorithm, using the password that is passed as an argument when creating an instance of the CryptoLayer class. + +Encrypting file contents protects against local access to these files, for example, protecting against hidden substitution of digital signatures of already known interlocutors. + +### 4.3. Digital signatures and data integrity + +For digital signing of packets, ECDSA (curve SECP256R1) is used. + +**The digital signature solves the MITM problem!** + +### 4.4. Key exchange + +For exchanging public keys, the ECDH protocol (curve SECP256R1) is used. Public keys are transmitted in the compressed X9.62 point format (for maximum efficient transmission over the communication channel). + +### 4.5. Masking + +To mask the transmission of bytes over the communication channel (messenger primarily), each byte is replaced with a specific word: before transmitting data to the module, each byte is replaced by a word from the dictionary. As a result, from a set of bytes `0x12 0x2 0x3f 0x4`, you get text `прямо лес пружина бег` (e.g., 'straight forest spring run'). This is done by the WordCoder component. + +## 5. Modules + +### 5.1. Where are existing modules located + +The official collection of ready-made modules for CryptoLayer is located [in this repository](https://github.com/igmunv/cryptolayer-modules). + +The collection is needed for using modules from it in applications that will use CryptoLayer. + +Your module can also be included there, just send a Pull Request and we will gladly accept it. + +### 5.2. Creating your own module + +#### 1. Preparation + +In a separate directory, create the files `main.py`, `requirements.txt`, `README.md`. + +In `requirements.txt`, you must specify all dependencies that the module uses. + +In `README.md`, provide a description of the module, what communication channel it uses, how it works, etc. + +#### 2. Import base_module + +In `main.py`, you need to import the library with the base class for modules: + +```python +from base_module import BaseModule, Credential +``` + +The `base_module` library is located [in this repository](https://github.com/igmunv/cryptolayer-module-interface). + +Don't worry if this library is not in your module's directory. When using CryptoLayer, application developers will add the necessary import dependencies and everything will work correctly. + +#### 3. Create a subclass of BaseModule + +Now you need to create a class that inherits from `BaseModule`: + +```python +... +class Example(BaseModule): +... +``` + +#### 4. Required fields of BaseModule + +Then, you need to implement the required fields: `name` (module name), `description` (module description), `unique_id` (unique module identifier): + +```python +class Example(BaseModule): +... +@property +def unique_id(self): return "ex.ample_1234" + +@property +def name(self): return "Example" + +@property +def description(self): return "Description for example" +... +``` + +#### 5. Login/Authorization Data (Credentials) + +Next, you need to specify the login/authorization data (Credentials). This may be needed, for example, if you are writing a module for a messenger or other service where authorization is required. This field can also be used to input other data, not just login details: port, IP address, keys, etc. - there are no limits. + +The login data will be requested by the application using CryptoLayer. + +To specify login data, the `expected_credentials` field is used - an array that must contain instances of the `Credential` class. The first constructor argument is the name, the second is a description of this data. As an example, we will ask the user to enter a login and password: + +```python +class Example(BaseModule): +... +expected_credentials = [Credential("Login", "User name, phone or email"), Credential("Password", "Password")] +... +``` + +If login data is not required, simply ignore the `expected_credentials` field. + +#### 6. Nested class Sender + +Next, you need to implement the nested Sender class, which is responsible for sending messages to the interlocutor's communication channel, specifically the send function, which is called by the transition pseudo-network layer of CryptoLayer. You don't need to change the function arguments; just implement sending the `text` argument to your communication channel. + +The `user_id` argument of the `__init__` function passes the user identifier in the communication channel. If an identifier is not required, do not use this field. + +The `credentials` argument of the `__init__` function passes the authorization data. They are in the same order as in the `expected_credentials` variable, but already in string format (list[str]). + +#### 7. Nested class Listener + +You also need to implement the second nested class Listener, which is responsible for receiving messages from the communication channel from the interlocutor, specifically the listen function, which should receive data from the interlocutor, and then must call the function located in the `ingester` field to pass the data up to the transition pseudo-network layer of CryptoLayer. + +```python +class Example(BaseModule): +... + class Listener: + ... + def listen(self) -> str: + ... + self.ingester(received_data) # Required, to pass data upwards + ... +... +``` + +The `user_id` argument of the `__init__` function passes the user identifier in the communication channel. If an identifier is not required, do not use this field. + +The `credentials` argument of the `__init__` function passes the authorization data. They are in the same order as in the `expected_credentials` variable, but already in string format (list[str]). + +#### 8. Function create_session + +The `create_session` function is called during CryptoLayer initialization (call to the `init` function). It is intended to create a session. In this function, instances of Sender and Listener are created, and dependencies for working with the communication channel are initialized (e.g., creating a session in the messenger). + +The `create_session` function takes one argument: `ingester` - a function is passed in this argument, which then must be passed to the Listener when creating an instance. + +You can override `__init__` for Sender and Listener if you need to pass other variables. The main thing is to pass `ingester` to the Listener, as without it, data will not be passed to CryptoLayer. + +### 5.3. Testing your own module + +To test your module, you can use [CryptoLayer CLI](https://github.com/igmunv/cryptolayer-cli). + +Download the contents of the [CryptoLayer CLI](https://github.com/igmunv/cryptolayer-cli) repository. + +Then run the program once using `./run.sh` or following the instructions in the [README.md](https://github.com/igmunv/cryptolayer-cli/blob/main/README.md). + +After that, exit the program. + +Copy your module's directory to `src/modules/`. + +Next, run CryptoLayer CLI, **BUT NOT via `./run.sh`, but as follows**: + +```bash +python3 -m venv venv +source venv/bin/activate +python3 src/modules/generate_reqs.py # Generate the list of module dependencies, including your new module +pip install -r src/modules/common_requirements.txt # Install module dependencies +python3 src/cryptolayer_cli.py +``` + +Your module will appear in CryptoLayer CLI, and now you can test it. + +If you need to change the module code, you can do it directly in `src/modules/`. + +**Just don't run `./run.sh` or the command `git submodule update --init --recursive` after copying the module to `src/modules/`, as this will delete your module!** + +After successful testing, you can send the module to [the official repository where CryptoLayer modules are collected](https://github.com/igmunv/cryptolayer-modules)! diff --git a/src/config.py b/src/config.py index 5705310..4abfbd4 100644 --- a/src/config.py +++ b/src/config.py @@ -7,6 +7,10 @@ LOGS_FILE_NAME = "crypto_layer.log" +# Длина node id: два uuid4 в шестнадцатеричном виде +NODE_ID_LENGTH = 64 + + # Размер чанка (в байтах) на транспортном уровне CHUNK_SIZE = 150 diff --git a/src/crypto_layer.py b/src/crypto_layer.py index 490b211..edb6f10 100644 --- a/src/crypto_layer.py +++ b/src/crypto_layer.py @@ -2,8 +2,10 @@ import sys import time import os +import re import uuid import logging +import hashlib from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives import serialization, hashes @@ -23,6 +25,10 @@ from UIProvider import UIProvider +# node id состоит только из шестнадцатеричных символов +NODE_ID_PATTERN = re.compile(r"\A[0-9a-f]+\Z") + + class CryptoLayer: @@ -79,6 +85,12 @@ def __init__(self, ui_provider: UIProvider, data_dir: str, module_class: BaseMod # ECC public key собеседника self.COMPANION_PUBLIC_KEY = None + # Выставляются при получении соответствующих данных от собеседника. + # Ожидающий поток просыпается сразу, а не на следующем тике опроса + self.COMPANION_NODE_ID_RECEIVED = threading.Event() + self.COMPANION_SIGN_RECEIVED = threading.Event() + self.COMPANION_PUBLIC_KEY_RECEIVED = threading.Event() + # Уровни self.TRANSITIONAL_LEVEL = None self.TRANSPORT_LEVEL = None @@ -110,6 +122,9 @@ def init(self): # удаление пароля из RAM self.remove_password_from_ram() + # Стек готов: только теперь имеет смысл проверять доступность собеседника + self.TRANSPORT_LEVEL.enable_ping() + self.ui_provider.on_ready() @@ -173,7 +188,7 @@ def generate_node_id(self): if os.path.exists(self.NODE_ID_FILE_PATH): node_id_file_content = open(self.NODE_ID_FILE_PATH, encoding="utf-8").read().strip() - if len(node_id_file_content) >= 64: + if self.check_node_id(node_id_file_content): self.NODE_ID = node_id_file_content return @@ -252,8 +267,7 @@ def node_id_exchange(self): self.APPLICATION_LEVEL.send_my_node_id(self.NODE_ID) self.ui_provider.update_status("Signatures", "Waiting for companion node id...", "in_progress") self.LOGGER.info("Signatures: Waiting for companion node id...") - while not self.COMPANION_NODE_ID: - time.sleep(0.1) + self.COMPANION_NODE_ID_RECEIVED.wait() self.ui_provider.update_status("Signatures", "Companion node id received!", "in_progress") self.LOGGER.info("Signatures: Companion node id received!") @@ -264,6 +278,11 @@ def check_and_exchange_companion_sign(self): # Отправка своей подписи # Ожидание подписи собеседника + # Станет True только если подпись собеседника действительно принята: + # либо совпала с ранее сохранённой, либо подтверждена пользователем. + # Управляет включением доверия в блоке finally ниже. + trust_ok = False + try: self.ui_provider.update_status("Signatures", "Send signature...", "in_progress") self.LOGGER.info("Signatures: Send signature...") @@ -271,13 +290,12 @@ def check_and_exchange_companion_sign(self): self.APPLICATION_LEVEL.send_my_sign(my_sign_public_bytes_X962) self.ui_provider.update_status("Signatures", "Waiting for companion signature...", "in_progress") self.LOGGER.info("Signatures: Waiting for companion signature...") - while not self.COMPANION_SIGN: - time.sleep(0.1) + self.COMPANION_SIGN_RECEIVED.wait() self.ui_provider.update_status("Signatures", "Companion signature received!", "in_progress") self.LOGGER.info("Signatures: Companion signature received!") # Затем сравнение с тем, что в файле - COMPANION_SIGN_FILE_PATH = os.path.join(self.KNOWN_NODES_DIR_PATH, self.COMPANION_NODE_ID) + COMPANION_SIGN_FILE_PATH = self.get_known_node_file_path(self.COMPANION_NODE_ID) if os.path.exists(COMPANION_SIGN_FILE_PATH): self.ui_provider.update_status("Signatures", "Сompanion signature exists", "in_progress") self.LOGGER.info("Signatures: Сompanion signature exists") @@ -294,6 +312,8 @@ def check_and_exchange_companion_sign(self): # Все норм они равны. Можем переходить к следующему этапу self.ui_provider.update_status("Signatures", "Companion signature exists", "in_progress") self.LOGGER.info("Signatures: Companion signature exists") + # Подпись совпала с ранее сохранённой для этого узла - доверяем ей. + trust_ok = True return else: @@ -314,7 +334,7 @@ def check_and_exchange_companion_sign(self): self.LOGGER.info("Signatures: user signatures check...") # Проверка подписи собеседника пользователем - if self.ui_provider.check_signatures(self.get_firts_last_4_chars_sign(self.SIGN_PUBLIC_KEY), self.get_firts_last_4_chars_sign(self.COMPANION_SIGN)): + if self.ui_provider.check_signatures(self.get_sign_fingerprint(self.SIGN_PUBLIC_KEY), self.get_sign_fingerprint(self.COMPANION_SIGN)): # Доверяем, записываем, используем эту подпись @@ -325,13 +345,17 @@ def check_and_exchange_companion_sign(self): comp_sign_public_bytes_X962 = self.get_key_bytes_X962(self.COMPANION_SIGN) self.encrypt_write_file(COMPANION_SIGN_FILE_PATH, self.USER_PASSWORD, comp_sign_public_bytes_X962) + # Пользователь сверил отпечаток и подтвердил его - доверяем подписи. + trust_ok = True + else: raise TypeError("do not trust the signature") finally: - self.TRANSITIONAL_LEVEL.COMPANION_SIGN_PUBLIC_KEY = self.COMPANION_SIGN # обновляем подпись собеседника - self.TRANSITIONAL_LEVEL.DO_SIGN = True # ОБЯЗАТЕЛЬНО!!! Так как теперь используется подпись + if trust_ok: + self.TRANSITIONAL_LEVEL.COMPANION_SIGN_PUBLIC_KEY = self.COMPANION_SIGN # обновляем подпись собеседника + self.TRANSITIONAL_LEVEL.DO_SIGN = True # ОБЯЗАТЕЛЬНО!!! Так как теперь используется подпись # Генерация и обмен публичными ключами, вычисление симметриного ключа @@ -347,6 +371,11 @@ def generate_and_exchange_ecc_keys(self): format=serialization.PublicFormat.CompressedPoint ) + # Всё, что пришло до включения обязательной проверки подписей, могло быть + # отправлено кем угодно. Отбрасываем такие данные и только после этого + # разрешаем приём ECDH-ключа собеседника + self.drop_unauthenticated_data() + # Передача публичного ключа # Ожидаем публичный ключ от собеседника self.ui_provider.update_status("Encryption", "Send public key...", "in_progress") @@ -355,8 +384,7 @@ def generate_and_exchange_ecc_keys(self): self.ui_provider.update_status("Encryption", "Waiting for companion public key...", "in_progress") self.LOGGER.info("Encryption: Waiting for companion public key...") - while not self.COMPANION_PUBLIC_KEY: - time.sleep(0.1) + self.COMPANION_PUBLIC_KEY_RECEIVED.wait() self.ui_provider.update_status("Encryption", "Companion public key received!", "in_progress") self.LOGGER.info("Encryption: Companion public key received!") @@ -364,7 +392,24 @@ def generate_and_exchange_ecc_keys(self): # Вычисление симетричного ключа self.ui_provider.update_status("Encryption", "Symmetric key computation...", "in_progress") self.LOGGER.info("Encryption: Symmetric key computation...") - self.AES_KEY = self.MY_PRIVATE_KEY.exchange(ec.ECDH(), self.COMPANION_PUBLIC_KEY) + shared_secret = self.MY_PRIVATE_KEY.exchange(ec.ECDH(), self.COMPANION_PUBLIC_KEY) + + # Результат ECDH - сам по себе он никак не привязан к тому, между кем + # шёл обмен. Поэтому не отдаём его в AES напрямую, а прогоняем через HKDF и + # подмешиваем в контекст оба публичных ключа и метку протокола. Порядок ключей + # фиксируем сортировкой, чтобы обе стороны независимо получили одинаковый + # контекст. В итоге ключ шифрования равномерный и жёстко связан именно с этой + # парой ключей и версией протокола: тот же секрет в другом контексте даст + # другой ключ, а переиспользовать секрет между сессиями не получится. + companion_public_key_bytes = self.get_key_bytes_X962(self.COMPANION_PUBLIC_KEY) + low_key, high_key = sorted((my_pkey_bytes, companion_public_key_bytes)) + hkdf = HKDF( + algorithm=hashes.SHA256(), + length=32, + salt=None, + info=b"cryptolayer ecdh-aes256 v1\x00" + low_key + high_key, + ) + self.AES_KEY = hkdf.derive(shared_secret) self.PRESENTATION_LEVEL.DO_ENCRYPT = True self.PRESENTATION_LEVEL.AES_KEY = self.AES_KEY @@ -372,13 +417,36 @@ def generate_and_exchange_ecc_keys(self): self.ui_provider.update_status("Encryption", "Done", "success") + # Отбросить всё, что пришло до включения обязательной проверки подписей, + # и разрешить приём ECDH-ключа собеседника. + # Вызывать только после DO_SIGN = True: данные, ещё не дошедшие до переходного + # уровня, будут проверены при обработке, поэтому чистим только то, что этот + # уровень уже успел пропустить. Уровни чистятся снизу вверх + def drop_unauthenticated_data(self): + + self.TRANSPORT_LEVEL.drop_pending_data() + self.PRESENTATION_LEVEL.take_pending_processing() + + # Буфер прикладного уровня чистится внутри expect_public_key под тем же + # замком, под которым выполняется его rworker + self.APPLICATION_LEVEL.expect_public_key() + + # Отправка сообщения def send(self, text): self.APPLICATION_LEVEL.send_text(text) def receive_node_id(self, node_id: str): + + # node id собеседника приходит по сети и используется как имя файла + # в known_nodes, поэтому принимается только node id ожидаемого вида + if not self.check_node_id(node_id): + self.LOGGER.error("companion node id is not valid: dropped") + return + self.COMPANION_NODE_ID = node_id + self.COMPANION_NODE_ID_RECEIVED.set() def receive_sign(self, sign: bytes): @@ -386,6 +454,7 @@ def receive_sign(self, sign: bytes): ec.SECP256R1(), sign ) + self.COMPANION_SIGN_RECEIVED.set() def receive_public_key(self, public_key: bytes): @@ -393,6 +462,7 @@ def receive_public_key(self, public_key: bytes): ec.SECP256R1(), public_key ) + self.COMPANION_PUBLIC_KEY_RECEIVED.set() def receive_text(self, timestamp: int, text: str): @@ -445,6 +515,23 @@ def decrypt_data_AES(self, data, password): return aesgcm.decrypt(nonce, encrypted_data, associated_data=None) + # Проверка, что node id имеет ожидаемый вид: два uuid4 в шестнадцатеричном виде + def check_node_id(self, node_id: str) -> bool: + return len(node_id) == config.NODE_ID_LENGTH and NODE_ID_PATTERN.match(node_id) is not None + + + # Путь к файлу с подписью узла внутри known_nodes. + # Дополнительная проверка на случай, если сюда попадёт непроверенный node id + def get_known_node_file_path(self, node_id: str) -> str: + + file_path = os.path.join(self.KNOWN_NODES_DIR_PATH, node_id) + + if os.path.dirname(os.path.realpath(file_path)) != os.path.realpath(self.KNOWN_NODES_DIR_PATH): + raise ValueError("node id points outside the known_nodes directory") + + return file_path + + def load_key_from_X962_bytes(self, key_bytes): return ec.EllipticCurvePublicKey.from_encoded_point( curve=ec.SECP256R1(), @@ -465,17 +552,20 @@ def remove_password_from_ram(self): self.USER_PASSWORD[i] = 0 - # Получить первые и последние 4 байта подписи - def get_firts_last_4_chars_sign(self, sign): + # Отпечаток публичного ключа подписи, который пользователи сверяют вручную по + # доверенному каналу. Это единственная защита от подмены ключей в момент + # установления доверия, поэтому отпечаток обязан зависеть от всего ключа целиком. + def get_sign_fingerprint(self, sign): - sign_public_bytes_pem = sign.public_bytes( + sign_public_bytes = sign.public_bytes( encoding=serialization.Encoding.X962, format=serialization.PublicFormat.CompressedPoint ) - first_4 = sign_public_bytes_pem[:4] - last_4 = sign_public_bytes_pem[-4:] - return f"{first_4.hex()}...{last_4.hex()}" + # Хэшируем ключ целиком через SHA-256 и показываем первые 160 бит + digest = hashlib.sha256(sign_public_bytes).digest()[:20] + hex_digest = digest.hex() + return " ".join(hex_digest[i:i + 4] for i in range(0, len(hex_digest), 4)) @@ -490,7 +580,8 @@ def stop(self, send_disconnect=True): # Ожидаем отправления всех пакетов timeout = 30 - while len(self.TRANSPORT_LEVEL.PENDING_ACK_PACKS) > 0 or len(self.APPLICATION_LEVEL.PENDING_SEND_BUF) > 0 or len(self.PRESENTATION_LEVEL.PENDING_SEND_BUF) > 0 or len(self.TRANSPORT_LEVEL.PENDING_SEND_BUF) > 0 or len(self.TRANSITIONAL_LEVEL.PENDING_SEND_BUF) > 0: + levels = (self.APPLICATION_LEVEL, self.PRESENTATION_LEVEL, self.TRANSPORT_LEVEL, self.TRANSITIONAL_LEVEL) + while self.TRANSPORT_LEVEL.PENDING_ACK_PACKS or any(level.PENDING_SEND_BUF.qsize() for level in levels): if timeout <= 0: break diff --git a/src/levels/application.py b/src/levels/application.py index 166e540..d40baff 100644 --- a/src/levels/application.py +++ b/src/levels/application.py @@ -1,14 +1,56 @@ import os +import threading import time +from enum import IntEnum + from levels.packet import ApplicationPacket, PackTypes, DataTypes, CMDTypes, TextMessagePacket from levels.base import Base +# Этапы рукопожатия. +# Служебные пакеты принимаются строго на своём этапе, всё остальное отбрасывается +class HandshakeStages(IntEnum): + + # Ждём node id собеседника + WAIT_NODE_ID = 0 + + # Ждём публичную часть подписи собеседника + WAIT_SIGN = 1 + + # Подпись собеседника получена, но пользователь ещё не подтвердил доверие к ней: + # обязательная проверка подписей на переходном уровне ещё не включена, + # поэтому доверять чему-либо пришедшему в этот момент нельзя + SIGN_RECEIVED = 2 + + # Проверка подписей включена, ждём ECDH-ключ собеседника + WAIT_PUBLIC_KEY = 3 + + # Рукопожатие завершено + READY = 4 + + class Application(Base): + def __init__(self): + + # Текущий этап рукопожатия. + # Выставляется до super(), так как Base сразу запускает потоки уровня + self.HANDSHAKE_STAGE = HandshakeStages.WAIT_NODE_ID + + # Этап меняется из потока инициализации (expect_public_key), а проверяется + # из потока уровня (rworker), поэтому обработка пакета и смена этапа + # не должны накладываться друг на друга + self.HANDSHAKE_STAGE_LOCK = threading.Lock() + + # Свой ECDH-ключ. Нужен, чтобы отправить его повторно (см. handle_packet) + self.MY_PUBLIC_KEY_BYTES = None + + super().__init__() + + def send_text(self, text: str): packet = ApplicationPacket(PackTypes.COMMUNIC.value, DataTypes.TEXT.value, TextMessagePacket(int(time.time()), text.encode()).to_bytes()) self.send(packet.to_bytes()) @@ -25,6 +67,7 @@ def send_my_sign(self, sign: bytes): def send_my_public_key(self, public_key: bytes): + self.MY_PUBLIC_KEY_BYTES = public_key packet = ApplicationPacket(PackTypes.SERVICE.value, CMDTypes.MY_PUBLIC_KEY.value, public_key) self.LOWER_LEVEL.send_without_encrypt(packet.to_bytes()) @@ -34,27 +77,82 @@ def send_disconnect(self): self.send(packet.to_bytes()) + # PUBLIC функция: её вызывает ядро, когда подпись собеседника проверена + # и включена обязательная проверка подписей у всех приходящих пакетов. + # Только с этого момента можно принимать ECDH-ключ собеседника + def expect_public_key(self): + with self.HANDSHAKE_STAGE_LOCK: + # Всё, что пришло до включения проверки подписей, доверия не заслуживает + self.take_pending_processing() + self.HANDSHAKE_STAGE = HandshakeStages.WAIT_PUBLIC_KEY + + + # Проверка, что пакет пришёл на своём этапе рукопожатия + def check_stage(self, expected_stage, packet_name): + + if self.HANDSHAKE_STAGE == expected_stage: + return True + + self.logger.warning(f"{packet_name} packet at stage {self.HANDSHAKE_STAGE.name}: dropped") + return False + + # постоянно читает данные из PENDING_PROCESSING_BUF и обрабатывает их и отправляет выше def rworker(self, data): + with self.HANDSHAKE_STAGE_LOCK: + self.handle_packet(data) + + + def handle_packet(self, data): packet = ApplicationPacket.from_bytes(data) if packet.pack_type == PackTypes.SERVICE.value: if packet.data_type == CMDTypes.MY_NODE_ID.value: + + if not self.check_stage(HandshakeStages.WAIT_NODE_ID, "MY_NODE_ID"): + return + + self.HANDSHAKE_STAGE = HandshakeStages.WAIT_SIGN self.UPPER_LEVEL.receive_node_id(packet.payload.decode()) elif packet.data_type == CMDTypes.MY_SIGN.value: + + if not self.check_stage(HandshakeStages.WAIT_SIGN, "MY_SIGN"): + return + + self.HANDSHAKE_STAGE = HandshakeStages.SIGN_RECEIVED self.UPPER_LEVEL.receive_sign(packet.payload) elif packet.data_type == CMDTypes.MY_PUBLIC_KEY.value: + + if self.HANDSHAKE_STAGE != HandshakeStages.WAIT_PUBLIC_KEY: + # Собеседник мог прислать ключ раньше, чем мы включили проверку подписей, + # или повторить отправку уже после рукопожатия - такой пакет не используется + self.logger.info(f"MY_PUBLIC_KEY packet at stage {self.HANDSHAKE_STAGE.name}: dropped") + return + + self.HANDSHAKE_STAGE = HandshakeStages.READY self.UPPER_LEVEL.receive_public_key(packet.payload) + # Наш ключ собеседник мог отбросить, если получил его до того, + # как включил у себя проверку подписей. Отправляем ещё раз + if self.MY_PUBLIC_KEY_BYTES is not None: + self.send_my_public_key(self.MY_PUBLIC_KEY_BYTES) + elif packet.data_type == CMDTypes.DISCONNECT.value: + + if not self.check_stage(HandshakeStages.READY, "DISCONNECT"): + return + self.UPPER_LEVEL.receive_disconnect() elif packet.pack_type == PackTypes.COMMUNIC.value: + if not self.check_stage(HandshakeStages.READY, "COMMUNIC"): + return + if packet.data_type == DataTypes.TEXT.value: text_packet = TextMessagePacket.from_bytes(packet.payload) self.UPPER_LEVEL.receive_text(text_packet.time, text_packet.payload.decode()) diff --git a/src/levels/base.py b/src/levels/base.py index 70da91d..26cab11 100644 --- a/src/levels/base.py +++ b/src/levels/base.py @@ -1,5 +1,5 @@ +import queue import threading -import time import logging @@ -10,15 +10,17 @@ class Base: # Здесь класс CryptoLayer. Это нужно для обратной связи от уровней core = None + # Как долго рабочий поток ждёт данные, прежде чем перепроверить stop_event. + # На задержку доставки не влияет: Queue будит поток сразу при put(). + POLL_TIMEOUT = 0.1 + def __init__(self): # Буффер пришедших данных - self.PENDING_PROCESSING_BUF = [] - self.PEND_PROC_BUF_LOCK = threading.Lock() + self.PENDING_PROCESSING_BUF = queue.Queue() # Буффер готовых к передаче данных - self.PENDING_SEND_BUF = [] - self.PEND_SEND_BUF_LOCK = threading.Lock() + self.PENDING_SEND_BUF = queue.Queue() self.logger = logging.getLogger(f"{self.__class__.__module__}.{self.__class__.__name__}") @@ -42,27 +44,48 @@ def update_levels(self, upper_level, lower_level): # PUBLIC фунция: её вызывает верхний уровень: отправь эти данные def send(self, data): self.logger.info(f"size: {len(data)}") - with self.PEND_SEND_BUF_LOCK: - self.PENDING_SEND_BUF.append(data) + self.PENDING_SEND_BUF.put(data) # PUBLIC фунция: её вызывает нижний уровень: получай эти данные def receive(self, data): self.logger.info(f"size: {len(data)}") - with self.PEND_PROC_BUF_LOCK: - self.PENDING_PROCESSING_BUF.append(data) + self.PENDING_PROCESSING_BUF.put(data) + + + # PUBLIC функция: забрать всё, что сейчас лежит в буффере приёма. + # Нужна, чтобы отбросить данные, пришедшие до того, как собеседник был проверен + def take_pending_processing(self): + + taken = [] + + while True: + try: + taken.append(self.PENDING_PROCESSING_BUF.get_nowait()) + except queue.Empty: + return taken + + + # Отдаёт рабочей функции всё, что попадает в буффер, пока не выставлен stop_event. + # Обработчик вызывается вне блокировки буффера: transport ждёт подтверждения + # секундами, и держать буффер занятым всё это время значит застопорить уровни выше. + def _pump(self, buffer, worker): + while not self.stop_event.is_set(): + try: + data = buffer.get(timeout=self.POLL_TIMEOUT) + except queue.Empty: + continue + self.logger.info(f"size: {len(data)}") + try: + worker(data) + except Exception: + # Иначе исключение убивает поток и уровень молча замолкает навсегда + self.logger.exception("worker failed, data dropped") # постоянно читает данные из PENDING_PROCESSING_BUF def receiver(self): - while not self.stop_event.is_set(): - with self.PEND_PROC_BUF_LOCK: - if self.PENDING_PROCESSING_BUF: - data = self.PENDING_PROCESSING_BUF[0] - self.logger.info(f"size: {len(data)}") - self.rworker(data) - del self.PENDING_PROCESSING_BUF[0] - time.sleep(0.1) + self._pump(self.PENDING_PROCESSING_BUF, self.rworker) # обрабатывает данные и отправляет выше @@ -72,17 +95,9 @@ def rworker(self, data): # постоянно читает PENDING_SEND_BUF def sender(self): - while not self.stop_event.is_set(): - with self.PEND_SEND_BUF_LOCK: - if self.PENDING_SEND_BUF: - data = self.PENDING_SEND_BUF[0] - self.logger.info(f"size: {len(data)}") - self.sworker(data) - del self.PENDING_SEND_BUF[0] - time.sleep(0.1) + self._pump(self.PENDING_SEND_BUF, self.sworker) + # формирует пакет и отправляет данные ниже def sworker(self, data): pass - - diff --git a/src/levels/transitional.py b/src/levels/transitional.py index 8c0b5d2..ba3b43d 100644 --- a/src/levels/transitional.py +++ b/src/levels/transitional.py @@ -57,6 +57,12 @@ def rworker(self, data): # постоянно читает PENDING_SEND_BUF, формирует пакет и отправляет данные ниже def sworker(self, data): + # Ключ подписи появляется только в signatures_setup, а пакеты нижних уровней + # (например, ping) могут попасть сюда раньше - подписать их нечем + if self.SIGN_PRIVATE_KEY is None: + self.logger.warning(f"signature key is not ready yet: packet dropped") + return + signature = self.SIGN_PRIVATE_KEY.sign( data, ec.ECDSA(hashes.SHA256()) diff --git a/src/levels/transport.py b/src/levels/transport.py index b0c975d..88e3512 100644 --- a/src/levels/transport.py +++ b/src/levels/transport.py @@ -2,12 +2,26 @@ import time import hashlib +from concurrent.futures import ThreadPoolExecutor + from levels.packet import TransportPacket from levels.base import Base class Transport(Base): + # Сколько ждать подтверждения перед повторной отправкой и сколько попыток всего. + # 6 x 5s держит прежний бюджет ~30s, но без бесконечной рекурсии + ACK_TIMEOUT = 5 + ACK_RETRIES = 6 + + # Сколько хешей уже обработанных пакетов помнить, чтобы отличить повтор + # от нового пакета. Должно перекрывать окно повторов ACK_TIMEOUT * ACK_RETRIES + SEEN_PACKETS_WINDOW = 512 + + # Сколько чанков одного потока держать в полёте одновременно + SEND_WINDOW = 8 + def __init__(self): super().__init__() @@ -24,12 +38,20 @@ def __init__(self): # ID потока: {count: количество пакетов в данном потоке, packets: [массив полученных пакетов в потоке]} self.WAITING_STREAMS = {} + # Хеши уже собранных пакетов, в порядке поступления. + # Читается и пишется только потоком receiver, поэтому без блокировки + self.SEEN_PACKETS = {} + # Текущий STREAM ID. Нужен для нумерации потоков байт self.CURRENT_STREAM_ID = 0 # Размер чанков данных в байтах self.CHUNK_SIZE = 100 + # Пинг включается ядром после инициализации. До этого стек ещё не готов + # отправлять пакеты, а само рукопожатие подтверждает, что собеседник на месте + self.PING_ENABLED = False + threading.Thread(target=self.every_second).start() @@ -38,7 +60,7 @@ def every_second(self): while not self.stop_event.is_set(): # Если больше 30 секунд от собеседника не приходило ни одного пакета, то отправляем пинг - if self.TIME_SINCE_LAST_PACKET > 30: + if self.PING_ENABLED and self.TIME_SINCE_LAST_PACKET > 30: self.send_with_pending_ping() # Прибавляем единицу, чтобы понимать сколько прошло секунд с получения последнего пакета @@ -48,6 +70,16 @@ def every_second(self): time.sleep(1) + # PUBLIC функция: её вызывает ядро, когда стек полностью проинициализирован. + # Счётчик обнуляется: рукопожатие только что прошло, собеседник точно на месте + def enable_ping(self): + + with self.TIME_SINCE_LAST_PACKET_LOCK: + self.TIME_SINCE_LAST_PACKET = 0 + + self.PING_ENABLED = True + + # Отправка ping, для проверки доступности собеседника, и ожидание ответа def send_with_pending_ping(self): @@ -102,27 +134,49 @@ def send_acknowledgment(self, rec_raw_packet_bytes): # Отправляем пакет и ожидаем подтверждение его получения def send_with_pending_acknowledgment(self, raw_packet_bytes, packet_hash): + # Event вместо опроса: поток просыпается в момент прихода подтверждения, + # а не на следующем тике таймера + acknowledged = threading.Event() with self.PENDING_ACK_PACKS_LOCK: - self.PENDING_ACK_PACKS[packet_hash] = 0 + self.PENDING_ACK_PACKS[packet_hash] = acknowledged - self.logger.info(f"send packet '{packet_hash}'") - self.LOWER_LEVEL.send(raw_packet_bytes) + try: + for attempt in range(self.ACK_RETRIES): - while packet_hash in self.PENDING_ACK_PACKS and not self.stop_event.is_set(): + self.logger.info(f"send packet '{packet_hash}' (attempt {attempt + 1})") + self.LOWER_LEVEL.send(raw_packet_bytes) - self.logger.info(f"wait ack...") + if acknowledged.wait(self.ACK_TIMEOUT): + self.logger.info(f"ack received!") + return - if self.PENDING_ACK_PACKS.get(packet_hash, 0) >= 30: - self.logger.warning(f"timeout while wait ack") - self.send_with_pending_acknowledgment(raw_packet_bytes, packet_hash) - return + if self.stop_event.is_set(): + return + self.logger.warning(f"timeout while wait ack, resending") + + self.logger.error(f"giving up on '{packet_hash}' after {self.ACK_RETRIES} attempts") + + finally: with self.PENDING_ACK_PACKS_LOCK: - self.PENDING_ACK_PACKS[packet_hash] = self.PENDING_ACK_PACKS[packet_hash] + 0.5 + self.PENDING_ACK_PACKS.pop(packet_hash, None) - time.sleep(0.5) - self.logger.info(f"ack received!") + # Видели ли уже в точности этот пакет. Запоминает его и вытесняет самый старый + def already_seen(self, raw_packet_bytes): + + hasher = hashlib.sha256() + hasher.update(raw_packet_bytes) + packet_hash = hasher.hexdigest() + + if packet_hash in self.SEEN_PACKETS: + return True + + self.SEEN_PACKETS[packet_hash] = None + if len(self.SEEN_PACKETS) > self.SEEN_PACKETS_WINDOW: + del self.SEEN_PACKETS[next(iter(self.SEEN_PACKETS))] + + return False # постоянно читает данные из PENDING_PROCESSING_BUF и обрабатывает их и отправляет выше @@ -162,30 +216,44 @@ def rworker(self, data): packet_hash = packet.payload.decode() with self.PENDING_ACK_PACKS_LOCK: - self.PENDING_ACK_PACKS.pop(packet_hash, None) + acknowledged = self.PENDING_ACK_PACKS.get(packet_hash) + # Снимает запись сам отправитель, здесь только будим его + if acknowledged: + acknowledged.set() # Если просто пакет передачи данных if packet.flags == 0x0: self.logger.info(f"data packet") - if packet.stream_id not in self.WAITING_STREAMS: - self.WAITING_STREAMS[packet.stream_id] = {"count": packet.chunk_count, "packets": []} - self.WAITING_STREAMS[packet.stream_id]["packets"].append({"chunk_id": packet.chunk_id, "payload": packet.payload}) + # Подтверждение отправляем всегда: отправитель повторяет пакет + # именно потому, что не увидел предыдущего подтверждения + self.send_acknowledgment(data) - self.logger.info(f"stream {packet.stream_id}: {len(self.WAITING_STREAMS[packet.stream_id]['packets'])} packet of {self.WAITING_STREAMS[packet.stream_id]['count']}") + # Повтор уже собранного пакета собирать заново нельзя: одночанковый + # поток доставился бы наверх дважды, а многочанковый склеился бы с чужим + if self.already_seen(data): + self.logger.info(f"duplicate packet, acknowledged and ignored") + return - # Отправка подтверждения о получении пакета - self.send_acknowledgment(data) + # Чанки лежат в словаре по chunk_id: повторно присланный чанк + # перезаписывает себя же, а не задваивает счётчик + stream = self.WAITING_STREAMS.setdefault( + packet.stream_id, {"count": packet.chunk_count, "packets": {}} + ) + stream["packets"][packet.chunk_id] = packet.payload + + self.logger.info(f"stream {packet.stream_id}: {len(stream['packets'])} packet of {stream['count']}") - if self.WAITING_STREAMS[packet.stream_id]["count"] == len(self.WAITING_STREAMS[packet.stream_id]["packets"]): + if stream["count"] == len(stream["packets"]): self.logger.info(f"all packets for this stream have been received!") - sorted_packets = sorted(self.WAITING_STREAMS[packet.stream_id]["packets"], key=lambda x: x["chunk_id"]) - data = bytes() - for _packet in sorted_packets: - data += _packet['payload'] + # Стрим удаляется сразу: иначе он копится в памяти, а после + # оборота stream_id через 256 новые чанки дописались бы в старый + del self.WAITING_STREAMS[packet.stream_id] + + data = b"".join(stream["packets"][chunk_id] for chunk_id in sorted(stream["packets"])) # Передаем выше self.UPPER_LEVEL.receive(data) @@ -197,10 +265,9 @@ def sworker(self, data): chunks = [data[i:i + self.CHUNK_SIZE] for i in range(0, len(data), self.CHUNK_SIZE)] self.logger.info(f"divided data into chunks: count: {len(chunks)}") + outgoing = [] for n, chunk in enumerate(chunks): - self.logger.info(f"start sending chunk {n}") - packet = TransportPacket(0x0, self.CURRENT_STREAM_ID, len(chunks), n, int(time.time()), chunk) raw_packet_bytes = packet.to_bytes() @@ -208,10 +275,40 @@ def sworker(self, data): hasher.update(raw_packet_bytes) packet_hash = hasher.hexdigest() - self.send_with_pending_acknowledgment(raw_packet_bytes, packet_hash) - self.logger.info(f"chunk sent!") + outgoing.append((raw_packet_bytes, packet_hash)) + + # Чанки летят с перекрытием: ожидание подтверждения по очереди стоило бы + # полного round-trip на каждый чанк. Получатель собирает поток по chunk_id, + # поэтому порядок прибытия внутри потока значения не имеет. + # Окно ограничено, чтобы большое сообщение не породило поток на каждый чанк + with ThreadPoolExecutor(max_workers=self.SEND_WINDOW) as pool: + futures = [pool.submit(self.send_with_pending_acknowledgment, raw, packet_hash) + for raw, packet_hash in outgoing] + for future in futures: + future.result() + + self.logger.info(f"all {len(chunks)} chunk(s) sent!") self.CURRENT_STREAM_ID = (self.CURRENT_STREAM_ID + 1) % 256 + # PUBLIC функция: её вызывает ядро. + # Отбросить уже принятые пакеты с данными и незавершённые потоки. + # Подтверждения и пинги не трогаем: на них завязана доставка + def drop_pending_data(self): + + for raw_packet_bytes in self.take_pending_processing(): + + try: + packet = TransportPacket.from_bytes(raw_packet_bytes) + except Exception as e: + self.logger.error(e) + continue + + if packet.flags != 0x0: + self.PENDING_PROCESSING_BUF.put(raw_packet_bytes) + + self.WAITING_STREAMS.clear() + + diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..85a3b00 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,237 @@ +"""Correctness proof for the CryptoLayer stack. + +Runs two real peers over an in-memory channel and asserts the messages that +come out are exactly the messages that went in -- across sizes, unicode, +multi-chunk streams, packet loss and stream-id wraparound. Also asserts the +plaintext never reaches the channel. + +Run: python3 tests/test_pipeline.py +""" +import logging +import os +import shutil +import sys +import tempfile +import threading +import time + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(ROOT, "src")) +sys.path.insert(0, os.path.join(ROOT, "bench")) + +from levels.transport import Transport # noqa: E402 +from loopback import Loopback # noqa: E402 +from UIProvider import UIProvider # noqa: E402 +from wordcoder import WordCoder # noqa: E402 + +SYLL_A = ["ba", "ve", "gi", "do", "zhu", "ki", "la", "mo", "ne", "pu", "ra", "so", "tu", "fi", "ha", "che"] +SYLL_B = ["lom", "ves", "gor", "dym", "zhar", "kit", "lug", "mox", "nos", "puh", "rov", "sud", "tir", "fon", "hor", "chan"] +WORDCODER_DICT = { + f"{a * 16 + b:02x}": SYLL_A[a] + SYLL_B[b] + for a in range(16) + for b in range(16) +} + +# Канал в памяти отвечает мгновенно, ждать продакшновые 5 секунд нечего +Transport.ACK_TIMEOUT = 0.4 + + +class CollectingUI(UIProvider): + def __init__(self): + self.ready = threading.Event() + self.received = [] + self.lock = threading.Lock() + + def request_data(self, prompt, data_type): + return "" + + def update_status(self, stage, message, status_type="in_progress"): + pass + + def on_text_received(self, timestamp, text): + with self.lock: + self.received.append(text) + + def check_signatures(self, my_sign, companion_sign): + return True + + def on_ready(self): + self.ready.set() + + def on_ping_timeout(self): + pass + + def on_disconnect(self): + pass + + +class Peers: + """Two handshaken CryptoLayer peers sharing an in-memory channel.""" + + def __init__(self, loss=0.0, record=False): + from crypto_layer import CryptoLayer + + self.root = tempfile.mkdtemp(prefix="cl-test-") + self.mod_a = Loopback(loss=loss, seed=1, record=record, label="A") + self.mod_b = Loopback(loss=loss, seed=2, record=record, label="B") + self.mod_a.peer_inbox = self.mod_b.inbox + self.mod_b.peer_inbox = self.mod_a.inbox + + self.ui_a, self.ui_b = CollectingUI(), CollectingUI() + self.a = CryptoLayer(self.ui_a, os.path.join(self.root, "a"), self.mod_a, "pw", WORDCODER_DICT) + self.b = CryptoLayer(self.ui_b, os.path.join(self.root, "b"), self.mod_b, "pw", WORDCODER_DICT) + + errors = [] + + def run(peer): + try: + peer.init() + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=run, args=(p,), daemon=True) for p in (self.a, self.b)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=120) + if errors: + raise errors[0] + assert self.ui_a.ready.is_set() and self.ui_b.ready.is_set(), "handshake did not finish" + + def exchange(self, messages, timeout=120): + """Send every message A->B and return what B actually received.""" + for text in messages: + self.a.send(text) + deadline = time.time() + timeout + while len(self.ui_b.received) < len(messages) and time.time() < deadline: + time.sleep(0.002) + return list(self.ui_b.received) + + def close(self): + from levels.base import Base + from base_module import BaseModule + + Base.stop_event.set() + BaseModule.stop_event.set() + time.sleep(0.3) + Base.stop_event.clear() + BaseModule.stop_event.clear() + shutil.rmtree(self.root, ignore_errors=True) + + +# ---------------------------------------------------------------- tests + +def test_wordcoder_roundtrip(): + wc = WordCoder(WORDCODER_DICT) + blob = bytes(range(256)) + os.urandom(500) + assert wc.decode(" ".join(wc.encode(blob)).split(" ")) == blob + + +def test_roundtrip_varied_sizes(): + """Every payload shape survives the full stack byte for byte.""" + messages = [ + "a", + "ok", + "Привет, как дела?", + "emoji and rtl: ok 123", + "x" * 99, # just under one chunk + "y" * 100, # exactly one chunk boundary + "z" * 101, # just over + "многочанковое " * 200, + "".join(chr(c) for c in range(32, 500)), + ] + peers = Peers() + try: + got = peers.exchange(messages) + assert got == messages, ( + f"payload mismatch: sent {len(messages)}, got {len(got)}; " + f"first diff at {next((i for i, (a, b) in enumerate(zip(messages, got)) if a != b), 'n/a')}" + ) + finally: + peers.close() + + +def test_multichunk_large_message(): + """A message spanning many transport chunks reassembles in order.""" + big = "".join(f"[{i:05d}]" for i in range(2000)) # ~14 KB, incompressible-ish counter + peers = Peers() + try: + got = peers.exchange([big]) + assert len(got) == 1, f"expected 1 message, got {len(got)}" + assert got[0] == big, "large message corrupted or reordered" + finally: + peers.close() + + +def test_survives_packet_loss(): + """With 30% of channel messages dropped, retransmission still delivers intact.""" + messages = [f"message number {i} " + "padding " * (i % 7) for i in range(15)] + peers = Peers(loss=0.30) + try: + got = peers.exchange(messages, timeout=180) + assert peers.mod_a.dropped + peers.mod_b.dropped > 0, "loss injection did not fire" + assert got == messages, f"loss corrupted the stream: got {len(got)}/{len(messages)}" + finally: + peers.close() + + +def test_stream_id_wraparound(): + """stream_id is one byte; more than 256 streams must not collide.""" + messages = [f"wrap {i}" for i in range(300)] + peers = Peers() + try: + got = peers.exchange(messages, timeout=180) + assert got == messages, ( + f"wraparound corrupted the stream at index " + f"{next((i for i, (a, b) in enumerate(zip(messages, got)) if a != b), len(got))}" + ) + finally: + peers.close() + + +def test_plaintext_never_hits_the_channel(): + """What the messenger sees must be words, and must not contain the secret.""" + secret = "SUPERSECRETCANARY9182" + peers = Peers(record=True) + try: + got = peers.exchange([secret]) + assert got == [secret] + wire = " ".join(peers.mod_a.wire) + assert wire, "nothing was recorded on the channel" + assert secret not in wire, "plaintext leaked onto the channel" + vocabulary = set(WORDCODER_DICT.values()) + tokens = set(wire.split(" ")) + assert tokens <= vocabulary, f"channel carried non-dictionary tokens: {tokens - vocabulary}" + finally: + peers.close() + + +TESTS = [ + test_wordcoder_roundtrip, + test_roundtrip_varied_sizes, + test_multichunk_large_message, + test_survives_packet_loss, + test_stream_id_wraparound, + test_plaintext_never_hits_the_channel, +] + + +def main(): + logging.disable(logging.CRITICAL) + failures = 0 + for test in TESTS: + name = test.__name__ + start = time.perf_counter() + try: + test() + except Exception as exc: + failures += 1 + print(f"FAIL {name} ({time.perf_counter() - start:.2f}s)\n {type(exc).__name__}: {exc}") + else: + print(f"ok {name} ({time.perf_counter() - start:.2f}s)") + print(f"\n{len(TESTS) - failures}/{len(TESTS)} passed") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main())