aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--CHECKPOINT.md2
-rw-r--r--Makefile2
-rw-r--r--README.md6
-rw-r--r--docs/research/086-sealed-sends.md96
-rw-r--r--research/experiments/014-sealed-sends/2026-09-27-drain-linux-aarch64.jsonl16
-rw-r--r--research/experiments/014-sealed-sends/drain.py124
-rw-r--r--serve/deliver.c42
-rw-r--r--serve/deliver.h4
-rw-r--r--serve/seal.c142
-rw-r--r--serve/seal.h32
-rw-r--r--serve/serve.c91
-rw-r--r--tests/check.py135
12 files changed, 676 insertions, 16 deletions
diff --git a/CHECKPOINT.md b/CHECKPOINT.md
index 2fe907c..3c948f6 100644
--- a/CHECKPOINT.md
+++ b/CHECKPOINT.md
@@ -15,7 +15,7 @@
| 앱 표현 | 상태(수, text와 커서·마크, 레코드의 리스트와 키 있는 리스트, 인덱스), 입력(키, text, 필드 있는 메시지, 타이머), 원자적 이벤트 핸들러, 순수 보기(조건식, 표와 걸러진 표, count·sum·max·min·first), `fn`·`use` 라이브러리, text 비교, 시간(`now`, `every`, `floor`, `%`). 로드할 때 타입·스택·합류·커서와 인덱스의 소속을 모두 검증하고 이벤트의 실행량에 상한이 있다 | 008, aup `docs/design/002-dsl-direction.md`, aup `make check`(파이썬 참조 모델 대조, 진단 132건) |
| 영속 | 선언한 상태가 곧 영속 상태다. 트리 저장소에 커밋당 동기화 한 번(헤드를 바이트 바로 앞에 두어 한 연속 구간), 이벤트 묶음마다 저장 한 번. `kill -9` 뒤 확인 응답한 이벤트가 모두 남는다. 레코드·키가 바뀐 모듈로 바꿔도 데이터가 이름으로 옮겨진다 | [076](docs/research/076-one-sync-commit.md), [081](docs/research/081-tail-records.md), 008 상태 영속·교체 |
| 인터페이스 | 한 명세가 TUI(셀 단위 갱신), CLI, 문장(NUI), 웹 GUI가 된다. 저장하면 실행 중인 앱이 상태를 이어 새 모듈로 바뀐다(`aup watch`, `vega … --watch`) | Vega `docs/design/001-first-component.md`, Vega `make check`(가상 터미널, 헤드리스 Chromium 포함) |
-| HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) |
+| HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). `--seal 출력=공개키`면 받는 노드의 키로 봉인해(HPKE) 사이의 중계가 읽지도 바꾸지도 못한다(봉인 전달 233k/s, 봉인 없이 470k/s). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md), [086](docs/research/086-sealed-sends.md) |
| 성능(Pi 5) | 문서 서버: 작은 GET 151.6k/s, 1 MiB GET 2,794/s(axum 1,372, drogon 708), 충돌 안전 PUT은 axum-atomic의 5.1–5.8배. 앱 서비스(aup 22줄 대 손으로 쓴 서비스): 내구성 있는 쓰기 약 76배(그 뒤 저장소 커밋을 한 구간으로 바꿔 64연결 21–26k/s, axum·drogon 내구 약 180/s), 20행 JSON GET 145k/s(axum 95k, drogon 144k), 쓰기와 섞인 읽기 149k/s·p99 0.6 ms(axum·drogon 내구 1.1–1.3k/s·p99 111–156 ms), 최대 RSS 3.0–4.1 MiB(axum 6.4, drogon 11.6) | [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) |
키로 나눈 소유자와 저장을 넘길 스레드는 하지 않기로 했다. 이 장치는 동기화를 한 줄로 세워, 나누면 쓰기가 23% 줄고 넘겨도
diff --git a/Makefile b/Makefile
index d9033a1..11d14d7 100644
--- a/Makefile
+++ b/Makefile
@@ -19,7 +19,7 @@ BUILD ?= build
INCLUDES = -Icore
MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c
STORE = store/store-tree.c
-SERVE = serve/serve.c serve/deliver.c serve/message-json.c
+SERVE = serve/serve.c serve/deliver.c serve/message-json.c serve/seal.c
HEADERS = $(wildcard core/*.h) $(wildcard serve/*.h)
OBJECTS = $(patsubst %.c,$(BUILD)/obj/%.o,$(MODEL) $(STORE))
diff --git a/README.md b/README.md
index 681c230..6ba38db 100644
--- a/README.md
+++ b/README.md
@@ -7,10 +7,12 @@ terminal UI, a CLI, spoken sentences and a web page through [Vega](/srv/git/vega
`pure-serve`: views are GETs and inputs are POSTs, with JSON bodies (groups included) read in place; answers are
durable, a commit being one contiguous write and one sync; views are read from a snapshot of the last save, so
reads never wait for a sync. What an app sends is saved with its event and delivered over HTTP in order and in
-batches, and a receiving app applies each message once. Measurements against axum and drogon on a Raspberry Pi 5
+batches, and a receiving app applies each message once; sealed to the receiver's key (HPKE), no relay between
+them can read or change it. Measurements against axum and drogon on a Raspberry Pi 5
are in [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md),
[080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md),
-[082](docs/research/082-json-in-place.md) and [083](docs/research/083-delivery-over-http.md); the status table is
+[082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) and
+[086](docs/research/086-sealed-sends.md); the status table is
[CHECKPOINT.md](CHECKPOINT.md).
The source follows the principle below, a small core with the rest selected around it
diff --git a/docs/research/086-sealed-sends.md b/docs/research/086-sealed-sends.md
new file mode 100644
index 0000000..08394b5
--- /dev/null
+++ b/docs/research/086-sealed-sends.md
@@ -0,0 +1,96 @@
+# 086 — 봉인한 전달: 중계가 읽지 못하는 앱 사이 메시지
+
+2026-09-27. [083](083-delivery-over-http.md)의 전달(앱의 `send`가 저장된 뒤 HTTP로 순서대로 묶여 가고, 받는 앱은 번호마다 한
+번 적용한다)에 종단 간 봉인을 더한 것. 로드맵의 E2EE 메신저로 가는 첫 단계다. 저장 형식과 전달 순서는 그대로다.
+
+착수 기준:
+
+- 바꿀 결정: 앱 사이 메시지를 TLS(구간마다 풀린다)에만 맡길지, 받는 노드의 공개키로 본문을 봉인해 그 사이의 중계·프록시·
+ 우편함이 읽지도 바꾸지도 못하게 할지. 방식을 새로 만들지 않는다. 표준 HPKE(RFC 9180, MLS·Oblivious HTTP·ECH가 쓰는 것)를
+ 쓴다.
+- 최소 확인: 기록하는 중계를 사이에 두고 모든 주문이 한 번씩 적용되며 중계가 항목도 필드 이름도 보지 못하는지, 한 바이트를
+ 바꾸면 거절되고 아무것도 적용되지 않는지, 기록한 요청을 다시 보내도 두 번 적용되지 않는지, 다른 키로 봉인한 것이 열리지
+ 않는지를 검사로 지킨다. 전달 속도를 봉인 전후로 번갈아 잰다.
+- 중단 기준: 봉인하지 않는 전달이 조금이라도 느려지면 되돌린다.
+
+## 한 것
+
+**봉인(`serve/seal.c`).** HPKE base 모드, X25519·HKDF-SHA256에 AEAD는 CPU에 AES 명령이 있으면 AES-128-GCM, 없으면
+ChaCha20-Poly1305다. OpenSSL 3.5의 `OSSL_HPKE_*`를 쓴다. 요청 하나의 본문(묶음이면 대괄호까지 한 덩어리)을 통째로 봉인하고,
+봉인은 그 요청의 출처·첫 번호·개수(`"source seq count"`)에 묶인다(AAD). 그래서 중계가 봉인한 본문을 다른 번호나 다른 출처로
+옮기거나 개수를 바꾸면 열리지 않는다. 본문은 캡슐화한 키 32 B, 암호문, 태그 16 B다. 헤더 `X-Pure-Seal`이 스위트를 적는다
+(`x25519-sha256-aes128gcm`이나 `x25519-sha256-chacha20poly1305`). 받는 쪽은 둘 다 연다. `Content-Type`은 연 뒤의 형식(JSON)을
+말한다. `Content-Encoding`이 gzip 위의 형식을 말하는 것과 같다.
+
+**키.** `--key FILE`은 원시 32 B 파일이다. 없으면 만들어 0600으로 쓰고 동기화한 뒤에 쓴다. 있으면 소유자만 읽을 수 있고 정확히
+32 B여야 한다(심볼릭 링크는 따라가지 않는다). 개인키는 OpenSSL의 보안 힙에 둔다. 메모리에 잠기고, 코어 덤프에 들지 않으며,
+풀 때 지워진다. 시작할 때 내는 JSON에 공개키(`"public_key"`, hex)가 있다.
+
+**보내는 쪽.** `--seal OUTPUT=KEY`는 `--send`로 연결한 출력의 요청을 받는 노드의 공개키로 봉인한다.
+
+**받는 쪽.** `X-Pure-Seal`이 있는 요청은 출처와 이 노드의 키가 있어야 하고 JSON이어야 한다(아니면 400 `seal`). 본문은 받은
+자리에서 열린다. 열리지 않으면(다른 키, 다른 요청의 묶임, 도중에 바뀜) 400 `seal`로 답하고 아무것도 적용하지 않는다. 연 평문은
+요청의 티켓이 반납될 때 0으로 지운다. `--sealed-only`인 노드는 봉인하지 않은 전달(`X-Pure-Source`가 있고 봉인이 없는 것)을
+403 `unsealed`로 거절한다.
+
+**한 덩어리로 보내기.** 봉인하려면 본문이 한 덩어리여야 해서, 먼저 요청을 버퍼 하나에 만들어 한 번에 쓰게 바꿨다. 전에는 묶음의
+보내기마다 쉼표나 대괄호 한 번, 본문 한 번을 썼다. 이것만으로 봉인하지 않는 전달이 91k → 470k 보내기/s가 됐다(아래). 머리를
+본문과 따로 한 번 더 써도 468k로 같았다. 이득은 본문 쪽이다. 버퍼 하나가 더 단순해서 그대로 뒀다. 이제 혼자 가는 보내기도 세그먼트
+둘이 아니라 하나로 간다.
+
+**검사(`tests/check.py`의 `server_sealed`).** 보내는 앱과 받는 앱 사이에 지나가는 요청을 모두 기록하는 파이썬 중계를 둔다.
+
+- 주문 60개가 모두 한 번씩 적용되고 합계가 맞는다.
+- 중계가 본 요청(25–26개) 모두에 `X-Pure-Seal`이 있고, 어느 본문에도 `secret`·`item`·`city`가 없다.
+- 기록한 요청의 가운데 한 비트를 바꿔 보내면 400 `seal`이다.
+- 원래 요청을 다시 보내면 200이지만 두 번 적용되지 않는다.
+- 봉인을 뺀 전달은 403 `unsealed`다.
+- 다른 노드의 키로 봉인하는 보내는 쪽의 요청은 중계를 지나도 열리지 않아 아무것도 적용되지 않는다.
+- 같은 키 파일로 다시 시작하면 공개키가 같다.
+
+이 검사를 새 디렉터리에서 여덟 번 연달아 돌려 모두 통과했다. 검사는 호스트가 포트를 스스로 고르게 한다(`--port 0`). 빈 포트를
+미리 알아낸 뒤 닫으면 그 사이에 다른 연결이 그 임시 포트를 가져갈 수 있다. 서버 검사 전체(`make check`)도 통과한다.
+
+## 장치가 말한 것
+
+Raspberry Pi 5(Cortex-A76 2.4 GHz), 루프백.
+
+**전달.** 받는 서버를 멈춘(`SIGSTOP`) 채 보내기 20,000개를 쌓고, 다시 움직이게 한 뒤 다 적용될 때까지 잰다. 두 서버는 코어
+0–2를 쓰고, 변형의 순서는 라운드마다 돌린다. 전달 요청은 어느 변형이든 80개다(묶음 하나에 최대 256개).
+([drain.py](../../research/experiments/014-sealed-sends/drain.py),
+[원시 결과](../../research/experiments/014-sealed-sends/2026-09-27-drain-linux-aarch64.jsonl).)
+
+| | 보내기/s (세 라운드) |
+|---|---:|
+| 전(보내기마다 두 번 씀, f2fd910) | 91.0–95.5k |
+| 한 덩어리, 봉인 없음 | **469.0–469.9k** |
+| 한 덩어리, 봉인 | **232.9–235.1k** |
+
+**기본 연산.** 코어 하나(`openssl speed`, 2초):
+
+| | 64 B | 1 KiB | 16 KiB |
+|---|---:|---:|---:|
+| AES-128-GCM | 183 MB/s | 1,350 MB/s | 2,282 MB/s |
+| ChaCha20-Poly1305 | 257 MB/s | 682 MB/s | 708 MB/s |
+
+X25519는 초당 6,105번(한 번 164 µs)이다. 묶음 본문(약 10 KiB)에서 AES-GCM이 세 배 빠르다. 혼자 가는 보내기(약 50 B)에서는
+ChaCha가 빠르지만 그 차이는 0.1 µs다. 봉인 비용의 대부분은 키 합의다. 요청 하나에 보내는 쪽이 두 번(임시 키 만들기, 합의),
+받는 쪽이 한 번, 모두 약 0.5 ms다. 한 전달선은 요청을 하나씩 보내고 답을 기다린다. 그래서 요청 80개면 약 40 ms가 된다. 봉인
+없이 20,000개가 걸리는 43 ms에 이것이 더해지면 83 ms, 240k/s로 잰 값과 맞는다. 대칭 암호에 드는 것은 요청마다 양쪽을 합쳐
+약 9 µs다.
+
+**봉인을 쓰지 않는 요청.** 헤더 하나와 분기 몇 개가 늘었다. 전(f2fd910)과 번갈아 네 라운드씩 쟀다(`benchmarks/http/run.py
+--baseline`). 앱의 합계 GET 64연결은 111,829 대 111,826/s, 문서의 64 B GET은 116,134 대 116,804/s이고, RSS는 같다.
+
+## 다음
+
+- **세션.** 전달선마다 캡슐화를 한 번만 하고, 그 뒤 요청은 HPKE 문맥 안의 번호로 봉인한다(RFC 9180의 여러 번 봉인). 캡슐화한
+ 키를 요청마다 헤더에 실으면 받는 쪽은 다시 시작한 뒤에도 따로 묻지 않고 열 수 있다. 키 합의가 세션당 한 번이 되어 봉인 전달이
+ 봉인하지 않은 전달에 가까워질 것이다.
+- **보낸 이 인증.** base 모드는 공개키를 아는 누구나 봉인할 수 있어서 `X-Pure-Source`가 인증되지 않는다. HPKE auth 모드로
+ 출처를 그 노드의 키에 묶고, `--sealed-only` 노드는 아는 보낸 이만 받게 한다.
+- **전방 비밀성.** 받는 노드의 키를 훔치면 기록해 둔 과거 전달이 열린다. 받는 키를 돌려 쓰거나(서명한 사전 키) 메신저에서는
+ 래칫을 쓴다.
+- **메타데이터.** 출처·번호·개수·경로·크기는 보인다. 크기는 채워 넣기로 가릴 수 있다.
+- **라우트 인증.** `--sealed-only`는 봉인하지 않은 전달을 거절할 뿐이다. 노드의 입력 라우트는 여전히 누구의 요청이든 받는다.
+ 앱 라우트의 인증은 따로 남은 일이다.
diff --git a/research/experiments/014-sealed-sends/2026-09-27-drain-linux-aarch64.jsonl b/research/experiments/014-sealed-sends/2026-09-27-drain-linux-aarch64.jsonl
new file mode 100644
index 0000000..f7a9477
--- /dev/null
+++ b/research/experiments/014-sealed-sends/2026-09-27-drain-linux-aarch64.jsonl
@@ -0,0 +1,16 @@
+{"round": 0, "variant": "plain", "sends": 20000, "sends_per_second": 469622, "receiver_requests": 85, "delivery_attempts": 80}
+{"round": 0, "variant": "sealed", "sends": 20000, "sends_per_second": 235074, "receiver_requests": 89, "delivery_attempts": 80}
+{"round": 0, "variant": "baseline", "sends": 20000, "sends_per_second": 91082, "receiver_requests": 102, "delivery_attempts": 80}
+{"round": 1, "variant": "sealed", "sends": 20000, "sends_per_second": 234780, "receiver_requests": 89, "delivery_attempts": 80}
+{"round": 1, "variant": "baseline", "sends": 20000, "sends_per_second": 90993, "receiver_requests": 102, "delivery_attempts": 80}
+{"round": 1, "variant": "plain", "sends": 20000, "sends_per_second": 469046, "receiver_requests": 85, "delivery_attempts": 80}
+{"round": 2, "variant": "baseline", "sends": 20000, "sends_per_second": 95491, "receiver_requests": 101, "delivery_attempts": 80}
+{"round": 2, "variant": "plain", "sends": 20000, "sends_per_second": 469907, "receiver_requests": 85, "delivery_attempts": 80}
+{"round": 2, "variant": "sealed", "sends": 20000, "sends_per_second": 232933, "receiver_requests": 89, "delivery_attempts": 80}
+{"aead": "aes-128-gcm", "bytes": 64, "kb_per_second": 182960.79}
+{"aead": "chacha20-poly1305", "bytes": 64, "kb_per_second": 257240.61}
+{"aead": "aes-128-gcm", "bytes": 1024, "kb_per_second": 1349597.18}
+{"aead": "chacha20-poly1305", "bytes": 1024, "kb_per_second": 682410.58}
+{"aead": "aes-128-gcm", "bytes": 16384, "kb_per_second": 2282373.12}
+{"aead": "chacha20-poly1305", "bytes": 16384, "kb_per_second": 707640.6}
+{"x25519_per_second": 6105.0}
diff --git a/research/experiments/014-sealed-sends/drain.py b/research/experiments/014-sealed-sends/drain.py
new file mode 100644
index 0000000..1a2819e
--- /dev/null
+++ b/research/experiments/014-sealed-sends/drain.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""How fast delivered sends drain, sealed and not (docs/research/086-sealed-sends.md).
+
+A sender app (tests/check.py's LINK_SENDER) takes --sends orders from eight
+clients while its receiver is stopped (SIGSTOP), so the sends pile up in the
+sender's store; the receiver is then continued and timed until it has
+applied them all. Runs plain and sealed (--seal to the receiver's --key) in
+alternating order per round, and a --baseline pure-serve (plain only) when
+given. Both servers on cores 0-2. Then the AEADs and X25519 on one core
+(openssl speed). Prints one JSON line per run.
+"""
+import argparse
+import json
+import os
+import select
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from http.client import HTTPConnection
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[2]
+sys.path.insert(0, str(ROOT / 'tests'))
+import check # noqa: E402
+
+
+def start(args):
+ process = subprocess.Popen(['taskset', '-c', '0-2'] + [str(a) for a in args] + ['--port', '0'],
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ ready, _, _ = select.select([process.stdout], [], [], 10)
+ line = process.stdout.readline() if ready else b''
+ assert line, process.communicate()[1][-1000:]
+ return process, json.loads(line)
+
+
+def count(port):
+ connection = HTTPConnection('127.0.0.1', port, timeout=5)
+ connection.request('GET', '/views/count')
+ value = int(connection.getresponse().read())
+ connection.close()
+ return value
+
+
+def run(work, binary, sends, sealed):
+ for name in ('a', 'b', 'key'):
+ shutil.rmtree(work / name, ignore_errors=True)
+ (work / 'key').mkdir()
+ key = ['--key', work / 'key' / 'receiver.key'] if sealed else []
+ receiver, started = start([binary, '--module', work / 'receiver.pmod', '--store', work / 'b', *key])
+ os.kill(receiver.pid, signal.SIGSTOP)
+ seal = ['--seal', f'ordered={started["public_key"]}'] if sealed else []
+ sender, sender_started = start([binary, '--module', work / 'sender.pmod', '--store', work / 'a',
+ '--send', f'ordered=http://127.0.0.1:{started["port"]}/inputs/order', *seal])
+
+ def client(first):
+ connection = HTTPConnection('127.0.0.1', sender_started['port'], timeout=30)
+ for i in range(first, sends, 8):
+ body = json.dumps({'item': f'i{i}', 'count': 1}).encode()
+ connection.request('POST', '/inputs/buy', body, {'Content-Type': 'application/json'})
+ connection.getresponse().read()
+ connection.close()
+
+ clients = [threading.Thread(target=client, args=(k,)) for k in range(8)]
+ for thread in clients:
+ thread.start()
+ for thread in clients:
+ thread.join()
+ began = time.perf_counter()
+ os.kill(receiver.pid, signal.SIGCONT)
+ while count(started['port']) < sends:
+ time.sleep(0.01)
+ elapsed = time.perf_counter() - began
+ stats = []
+ for process in (sender, receiver):
+ process.terminate()
+ _, error = process.communicate()
+ stats.append([json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1])
+ return {'sends_per_second': round(sends / elapsed), 'receiver_requests': stats[1]['requests'],
+ 'delivery_attempts': stats[0]['delivery_attempts']}
+
+
+def speeds():
+ for size in (64, 1024, 16384):
+ for cipher in ('aes-128-gcm', 'chacha20-poly1305'):
+ out = subprocess.run(['taskset', '-c', '3', 'openssl', 'speed', '-seconds', '2', '-bytes', str(size), '-evp',
+ cipher], capture_output=True, text=True, check=True).stdout.splitlines()[-1]
+ yield {'aead': cipher, 'bytes': size, 'kb_per_second': float(out.split()[-1].rstrip('k'))}
+ out = subprocess.run(['taskset', '-c', '3', 'openssl', 'speed', '-seconds', '2', 'ecdhx25519'],
+ capture_output=True, text=True, check=True).stdout.splitlines()[-1]
+ yield {'x25519_per_second': float(out.split()[-1])}
+
+
+def main():
+ parser = argparse.ArgumentParser()
+ parser.add_argument('--aup', type=Path, default=ROOT.parent / 'aup' / 'build' / 'aup')
+ parser.add_argument('--binary', type=Path, default=ROOT / 'build' / 'pure-serve')
+ parser.add_argument('--baseline', type=Path, help='an earlier pure-serve, run plain')
+ parser.add_argument('--sends', type=int, default=20000)
+ parser.add_argument('--rounds', type=int, default=3)
+ args = parser.parse_args()
+ with tempfile.TemporaryDirectory(prefix='pure-drain-') as directory:
+ work = Path(directory)
+ for name, text in (('sender', check.LINK_SENDER), ('receiver', check.LINK_RECEIVER)):
+ (work / f'{name}.aup').write_text(text)
+ check.compile_module(args.aup.resolve(), work / f'{name}.aup', work / f'{name}.pmod')
+ for round_ in range(args.rounds):
+ variants = [('plain', args.binary, False), ('sealed', args.binary, True)]
+ if args.baseline:
+ variants.append(('baseline', args.baseline.resolve(), False))
+ shift = round_ % len(variants)
+ for name, binary, sealed in variants[shift:] + variants[:shift]:
+ result = run(work, binary, args.sends, sealed)
+ print(json.dumps({'round': round_, 'variant': name, 'sends': args.sends, **result}), flush=True)
+ for line in speeds():
+ print(json.dumps(line), flush=True)
+
+
+if __name__ == '__main__':
+ main()
diff --git a/serve/deliver.c b/serve/deliver.c
index fd0cb66..391e0db 100644
--- a/serve/deliver.c
+++ b/serve/deliver.c
@@ -1,5 +1,6 @@
#define _POSIX_C_SOURCE 200809L
#include "deliver.h"
+#include "seal.h"
#include "message-json.h"
#include "module-text.h"
#include <errno.h>
@@ -50,6 +51,8 @@ struct PureWire {
size_t answer_head, answer_count, answer_capacity;
Answer taken; /* the one the owner took last, kept until it takes the next */
uint32_t placed[2 * PURE_MODULE_FIELDS];
+ bool sealed; /* the sends go sealed to peer's key */
+ unsigned char peer[PURE_SEAL_KEY];
};
/* ---- TLS: one client context, verifying the receiver's certificate against
@@ -554,33 +557,56 @@ static bool form_answer(const PureWire *w, int status, const Text *body, const P
static int send_request(PureWire *w, const char *request, size_t size, const Pending *p, Answer *answer);
+void pure_wire_seal(PureWire *w, const unsigned char peer[PURE_SEAL_KEY]) {
+ memcpy(w->peer, peer, PURE_SEAL_KEY);
+ w->sealed = true;
+}
+
/* One request for count sends from p (numbered in a row): one object, or
* an array of them for a receiver that takes batches. A final answer to a
* send whose answers are wired is made into *answer. */
static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) {
size_t body = count > 1 ? count + 1 : 0;
for (unsigned i = 0; i < count; ++i) body += p[i].size;
- char batch[32] = "";
+ size_t length = body + (w->sealed ? PURE_SEAL_OVERHEAD : 0);
+ char batch[32] = "", seal[48] = "";
if (count > 1) snprintf(batch, sizeof(batch), "X-Pure-Count: %u\r\n", count);
+ if (w->sealed) snprintf(seal, sizeof(seal), "X-Pure-Seal: %s\r\n", pure_seal_suite());
/* The request in one piece, its head and then a batch in its brackets:
* one write where there were two a send (a batch drained at 91k sends a
- * second, and at 470k in one piece). */
+ * second, and at 470k in one piece). Sealed, the body is sealed whole,
+ * bound to its source, first number and count. */
enum { HEAD = 2048 }; /* the longest path, host and source leave room to spare */
- char *request = malloc(HEAD + body);
- if (!request) return AGAIN;
+ char *request = malloc(HEAD + length);
+ unsigned char *plain = w->sealed ? malloc(body) : NULL;
+ if (!request || (w->sealed && !plain)) {
+ free(request);
+ free(plain);
+ return AGAIN;
+ }
int head = snprintf(request, HEAD,
"POST %s HTTP/1.1\r\nHost: %s:%s\r\nContent-Type: application/json\r\nContent-Length: %zu\r\n"
- "X-Pure-Source: %s\r\nX-Pure-Sequence: %llu\r\n%sIdempotency-Key: %s-%llu\r\n\r\n",
- w->path, w->host, w->port, body, w->source, (unsigned long long)p->seq, batch, w->source,
+ "X-Pure-Source: %s\r\nX-Pure-Sequence: %llu\r\n%s%sIdempotency-Key: %s-%llu\r\n\r\n",
+ w->path, w->host, w->port, length, w->source, (unsigned long long)p->seq, batch, seal, w->source,
(unsigned long long)p->seq);
- char *at = request + head;
+ unsigned char *at = plain ? plain : (unsigned char *)request + head;
for (unsigned i = 0; i < count; ++i) {
if (count > 1) *at++ = i ? ',' : '[';
memcpy(at, p[i].body, p[i].size);
at += p[i].size;
}
if (count > 1) *at++ = ']';
- int result = send_request(w, request, (size_t)(at - request), p, answer);
+ if (plain) {
+ char aad[PURE_SOURCE + 48];
+ snprintf(aad, sizeof(aad), "%s %llu %u", w->source, (unsigned long long)p->seq, count);
+ bool sealed = pure_seal(w->peer, aad, plain, body, (unsigned char *)request + head);
+ free(plain);
+ if (!sealed) {
+ free(request);
+ return AGAIN;
+ }
+ }
+ int result = send_request(w, request, (size_t)head + length, p, answer);
free(request);
return result;
}
diff --git a/serve/deliver.h b/serve/deliver.h
index 2d1da29..70b03b9 100644
--- a/serve/deliver.h
+++ b/serve/deliver.h
@@ -23,6 +23,10 @@ void pure_wire_trust(const char *ca_file);
PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source,
char *error, size_t size);
uint32_t pure_wire_output(const PureWire *);
+/* Seal the wire's sends to the receiver's public key (serve/seal.h): each
+ * request's body, bound to its source, first number and count, goes with
+ * X-Pure-Seal naming the suite. Before pure_wire_start. */
+void pure_wire_seal(PureWire *, const unsigned char peer[32]);
/* The owner's side: queue a saved send numbered after the last queued one
* (it is encoded now); false when it cannot be (a bytes field that is not
* UTF-8 has no JSON form), and the wire then stops at it. */
diff --git a/serve/seal.c b/serve/seal.c
new file mode 100644
index 0000000..c1cc55b
--- /dev/null
+++ b/serve/seal.c
@@ -0,0 +1,142 @@
+#define _GNU_SOURCE
+#include "seal.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <openssl/crypto.h>
+#include <openssl/evp.h>
+#include <openssl/hpke.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#if defined(__aarch64__) && defined(__linux__)
+#include <asm/hwcap.h>
+#include <sys/auxv.h>
+#endif
+
+struct PureSealKey { EVP_PKEY *key; };
+
+static const unsigned char info[] = "pure sealed send 1";
+static const char aes_name[] = "x25519-sha256-aes128gcm", chacha_name[] = "x25519-sha256-chacha20poly1305";
+
+/* AES-GCM where AES is in the CPU (two to three times ChaCha20-Poly1305 from
+ * a kilobyte up on a Cortex-A76, research 086); ChaCha20-Poly1305, fast in
+ * plain code, elsewhere. */
+static bool has_aes(void) {
+#if defined(__aarch64__) && defined(__linux__)
+ return (getauxval(AT_HWCAP) & HWCAP_AES) && (getauxval(AT_HWCAP) & HWCAP_PMULL);
+#elif defined(__x86_64__) && (defined(__GNUC__) || defined(__clang__))
+ return __builtin_cpu_supports("aes") && __builtin_cpu_supports("pclmul");
+#else
+ return false;
+#endif
+}
+
+static bool suite_named(const char *name, OSSL_HPKE_SUITE *out) {
+ OSSL_HPKE_SUITE suite = {OSSL_HPKE_KEM_ID_X25519, OSSL_HPKE_KDF_ID_HKDF_SHA256, 0};
+ if (!strcmp(name, aes_name)) suite.aead_id = OSSL_HPKE_AEAD_ID_AES_GCM_128;
+ else if (!strcmp(name, chacha_name)) suite.aead_id = OSSL_HPKE_AEAD_ID_CHACHA_POLY1305;
+ else return false;
+ *out = suite;
+ return true;
+}
+
+const char *pure_seal_suite(void) { return has_aes() ? aes_name : chacha_name; }
+
+static pthread_once_t secure_once = PTHREAD_ONCE_INIT;
+static void secure_heap(void) { CRYPTO_secure_malloc_init(16384, 32); }
+
+static bool hex_value(char c, unsigned *v) {
+ if (c >= '0' && c <= '9') *v = (unsigned)(c - '0');
+ else if (c >= 'a' && c <= 'f') *v = (unsigned)(c - 'a' + 10);
+ else if (c >= 'A' && c <= 'F') *v = (unsigned)(c - 'A' + 10);
+ else return false;
+ return true;
+}
+
+bool pure_seal_peer(const char *hex, unsigned char out[PURE_SEAL_KEY]) {
+ if (strlen(hex) != 2 * PURE_SEAL_KEY) return false;
+ for (size_t i = 0; i < PURE_SEAL_KEY; ++i) {
+ unsigned high, low;
+ if (!hex_value(hex[2 * i], &high) || !hex_value(hex[2 * i + 1], &low)) return false;
+ out[i] = (unsigned char)(high << 4 | low);
+ }
+ return true;
+}
+
+PureSealKey *pure_seal_key(const char *path, char public_hex[2 * PURE_SEAL_KEY + 1], char *error, size_t size) {
+ pthread_once(&secure_once, secure_heap);
+ unsigned char *raw = OPENSSL_secure_malloc(PURE_SEAL_KEY), public_key[PURE_SEAL_KEY];
+ PureSealKey *made = calloc(1, sizeof(*made));
+ EVP_PKEY *key = NULL;
+ size_t length = PURE_SEAL_KEY;
+ if (!raw || !made) { snprintf(error, size, "out of memory"); goto fail; }
+ int fd = open(path, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+ struct stat info_;
+ if (fd >= 0) {
+ bool ok = !fstat(fd, &info_) && S_ISREG(info_.st_mode) && !(info_.st_mode & 077) &&
+ read(fd, raw, PURE_SEAL_KEY) == PURE_SEAL_KEY && info_.st_size == PURE_SEAL_KEY;
+ close(fd);
+ if (!ok) { snprintf(error, size, "%s: a key is 32 bytes that only its owner may read (0600)", path); goto fail; }
+ key = EVP_PKEY_new_raw_private_key(EVP_PKEY_X25519, NULL, raw, PURE_SEAL_KEY);
+ } else if (errno == ENOENT) {
+ /* A new key: made, written whole and synced before it is used. */
+ key = EVP_PKEY_Q_keygen(NULL, NULL, "X25519");
+ fd = open(path, O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600);
+ bool ok = key && fd >= 0 && EVP_PKEY_get_raw_private_key(key, raw, &length) == 1 && length == PURE_SEAL_KEY &&
+ write(fd, raw, PURE_SEAL_KEY) == PURE_SEAL_KEY && !fsync(fd);
+ if (fd >= 0 && close(fd)) ok = false;
+ if (!ok) { snprintf(error, size, "%s: cannot make a key there", path); unlink(path); goto fail; }
+ } else { snprintf(error, size, "%s: %s", path, strerror(errno)); goto fail; }
+ length = PURE_SEAL_KEY;
+ if (!key || EVP_PKEY_get_raw_public_key(key, public_key, &length) != 1 || length != PURE_SEAL_KEY) {
+ snprintf(error, size, "%s: not an X25519 key", path);
+ goto fail;
+ }
+ for (size_t i = 0; i < PURE_SEAL_KEY; ++i) snprintf(public_hex + 2 * i, 3, "%02x", public_key[i]);
+ OPENSSL_secure_clear_free(raw, PURE_SEAL_KEY);
+ made->key = key;
+ return made;
+fail:
+ EVP_PKEY_free(key);
+ if (raw) OPENSSL_secure_clear_free(raw, PURE_SEAL_KEY);
+ free(made);
+ return NULL;
+}
+
+void pure_seal_key_free(PureSealKey *key) {
+ if (!key) return;
+ EVP_PKEY_free(key->key);
+ free(key);
+}
+
+bool pure_seal(const unsigned char peer[PURE_SEAL_KEY], const char *aad, const unsigned char *plain, size_t size,
+ unsigned char *out) {
+ OSSL_HPKE_SUITE suite;
+ if (!suite_named(pure_seal_suite(), &suite)) return false;
+ OSSL_HPKE_CTX *context = OSSL_HPKE_CTX_new(OSSL_HPKE_MODE_BASE, suite, OSSL_HPKE_ROLE_SENDER, NULL, NULL);
+ size_t encapsulated = PURE_SEAL_KEY, sealed = size + 16;
+ bool ok = context && OSSL_HPKE_encap(context, out, &encapsulated, peer, PURE_SEAL_KEY, info, sizeof(info) - 1) == 1 &&
+ encapsulated == PURE_SEAL_KEY &&
+ OSSL_HPKE_seal(context, out + PURE_SEAL_KEY, &sealed, (const unsigned char *)aad, strlen(aad), plain, size) == 1 &&
+ sealed == size + 16;
+ OSSL_HPKE_CTX_free(context);
+ return ok;
+}
+
+bool pure_unseal(const PureSealKey *key, const char *suite_name, const char *aad, const unsigned char *sealed, size_t size,
+ unsigned char *plain) {
+ OSSL_HPKE_SUITE suite;
+ if (!key || size < PURE_SEAL_OVERHEAD || !suite_named(suite_name, &suite)) return false;
+ OSSL_HPKE_CTX *context = OSSL_HPKE_CTX_new(OSSL_HPKE_MODE_BASE, suite, OSSL_HPKE_ROLE_RECEIVER, NULL, NULL);
+ size_t opened = size - PURE_SEAL_OVERHEAD;
+ bool ok = context &&
+ OSSL_HPKE_decap(context, sealed, PURE_SEAL_KEY, key->key, info, sizeof(info) - 1) == 1 &&
+ OSSL_HPKE_open(context, plain, &opened, (const unsigned char *)aad, strlen(aad), sealed + PURE_SEAL_KEY,
+ size - PURE_SEAL_KEY) == 1 &&
+ opened == size - PURE_SEAL_OVERHEAD;
+ OSSL_HPKE_CTX_free(context);
+ return ok;
+}
diff --git a/serve/seal.h b/serve/seal.h
new file mode 100644
index 0000000..4ce3844
--- /dev/null
+++ b/serve/seal.h
@@ -0,0 +1,32 @@
+#ifndef PURE_SEAL_H
+#define PURE_SEAL_H
+#include <stdbool.h>
+#include <stddef.h>
+
+/* Sends sealed end to end (docs/research/086-sealed-sends.md): a receiving
+ * node's key opens what a sender sealed to its public key, and nothing
+ * between them (a relay, a proxy, a mailbox) can read or change it. HPKE
+ * (RFC 9180) base mode: X25519, HKDF-SHA256, and AES-128-GCM where the CPU
+ * has AES instructions, else ChaCha20-Poly1305. A private key lives in
+ * OpenSSL's secure heap: locked in memory, out of core dumps, cleared when
+ * freed. */
+enum { PURE_SEAL_KEY = 32, PURE_SEAL_OVERHEAD = 32 + 16 }; /* the encapsulated key, then the tag */
+
+typedef struct PureSealKey PureSealKey;
+/* A node's key: read from path, or made and written there (0600) when there
+ * is none; its public half in hex. NULL with why in error. */
+PureSealKey *pure_seal_key(const char *path, char public_hex[2 * PURE_SEAL_KEY + 1], char *error, size_t size);
+void pure_seal_key_free(PureSealKey *);
+/* A public key from its hex. */
+bool pure_seal_peer(const char *hex, unsigned char out[PURE_SEAL_KEY]);
+/* The suite this machine seals with, as the X-Pure-Seal header names it. */
+const char *pure_seal_suite(void);
+/* plain sealed to peer, bound to aad: size + PURE_SEAL_OVERHEAD bytes into out. */
+bool pure_seal(const unsigned char peer[PURE_SEAL_KEY], const char *aad, const unsigned char *plain, size_t size,
+ unsigned char *out);
+/* Opened with key under suite (either this machine's or the other known one):
+ * size - PURE_SEAL_OVERHEAD bytes into plain; false when it does not open
+ * (another key, another aad, or changed on the way). */
+bool pure_unseal(const PureSealKey *, const char *suite, const char *aad, const unsigned char *sealed, size_t size,
+ unsigned char *plain);
+#endif
diff --git a/serve/serve.c b/serve/serve.c
index 2ea7adb..8f0bb8d 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -30,6 +30,7 @@
* threads hand it their ready requests and answer what it hands back. Its
* bursts take the requests of every thread, one save each. */
#include "deliver.h"
+#include "seal.h"
#include "message-json.h"
#include "module-store.h"
#include "module-text.h"
@@ -69,6 +70,11 @@ typedef struct Host Host;
typedef enum { ROUTE_NONE, ROUTE_OBJECT, ROUTE_VIEW, ROUTE_INPUT } RouteKind;
typedef struct { RouteKind kind; uint32_t target; size_t from, count; } Route;
+/* This node's key, for deliveries sealed to it (--key), and whether it
+ * takes only those (--sealed-only). */
+static PureSealKey *seal_key;
+static bool sealed_only;
+
/* Allocated when accepted and reused after closing: memory follows the
* connections that are open, not the configured maximum. */
typedef struct Connection {
@@ -94,6 +100,11 @@ typedef struct Connection {
char source[PURE_SOURCE + 1]; /* a delivered message's sender (X-Pure-Source), or empty */
uint64_t sequence; /* and its number (X-Pure-Sequence) */
unsigned batch; /* how many numbered messages the body holds (X-Pure-Count) */
+ /* A sealed delivery: the suite it names (X-Pure-Seal), what the seal is
+ * bound to (source, number, count), and how many plain bytes the body
+ * holds once opened, cleared when its ticket goes. */
+ char seal[40], aad[PURE_SOURCE + 48];
+ size_t opened;
int64_t expected;
size_t body_size, received, in_used;
unsigned char *body;
@@ -247,6 +258,8 @@ static bool number(const char *text, uint64_t maximum, uint64_t *value) {
static void release_ticket(void *owner) {
Connection *c = owner;
+ if (c->opened && c->body) explicit_bzero(c->body, c->opened); /* an opened seal's plain text is gone with it */
+ c->opened = 0;
if (c->ticket.runtime && !pure_runtime_release(c->ticket)) abort();
c->ticket = (PureTicket){0};
c->body = NULL;
@@ -413,6 +426,7 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
c->source[0] = 0;
c->sequence = 0;
c->batch = 1;
+ c->seal[0] = 0;
bool counted = false;
c->body_size = 0;
c->json = false;
@@ -464,6 +478,10 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
} else if (!strcasecmp(line, "X-Pure-Sequence")) {
if (c->sequence || !number(value, UINT64_MAX, &n) || !n) return false;
c->sequence = n;
+ } else if (!strcasecmp(line, "X-Pure-Seal")) {
+ /* A delivery sealed to this node's key, in the suite named. */
+ if (*c->seal || strlen(value) >= sizeof(c->seal)) return false;
+ strcpy(c->seal, value);
} else if (!strcasecmp(line, "X-Pure-Count")) {
/* Numbered messages from X-Pure-Sequence on, as a JSON array. */
if (counted || !number(value, BATCH, &n) || !n) return false;
@@ -600,9 +618,31 @@ static const char *convert(Connection *c) {
return NULL;
}
+/* A sealed body opened where it lies, or false (another key, another
+ * delivery's binding, or changed on the way). */
+static bool open_seal(Connection *c) {
+ if (c->body_size < PURE_SEAL_OVERHEAD) return false;
+ size_t size = c->body_size - PURE_SEAL_OVERHEAD;
+ unsigned char *plain = malloc(size ? size : 1);
+ bool opened = plain && pure_unseal(seal_key, c->seal, c->aad, c->body, c->body_size, plain);
+ if (opened) {
+ memcpy(c->body, plain, size);
+ c->body_size = size;
+ c->opened = size;
+ }
+ if (plain) explicit_bzero(plain, size);
+ free(plain);
+ return opened;
+}
+
/* A request's body is in: an app input's JSON becomes its payload, and the
* request is ready, or it is answered 400 and never runs. */
static void body_done(Connection *c) {
+ if (*c->seal && !open_seal(c)) {
+ release_ticket(c); /* never made ready: it never runs */
+ respond(c, 400, "Bad Request", "seal", 0, (PureSpan){0});
+ return;
+ }
const char *why = c->route == ROUTE_INPUT && c->json ? convert(c) : NULL;
if (why) {
release_ticket(c); /* never made ready: it never runs */
@@ -623,6 +663,13 @@ static void start_request(Connection *c, size_t head_size) {
memmove(c->in, c->in + head_size, extra);
c->in_used = extra;
if (!parse_head(c, head, head_size, &query)) { refuse(c, 400, "Bad Request", "request"); return; }
+ /* A node that keeps its deliveries sealed refuses the others; a seal
+ * needs this node's key, and binds the delivery's source, number and count. */
+ if (*c->source && sealed_only && !*c->seal) { refuse(c, 403, "Forbidden", "unsealed"); return; }
+ if (*c->seal) {
+ if (!*c->source || !seal_key || !c->json) { refuse(c, 400, "Bad Request", "seal"); return; }
+ snprintf(c->aad, sizeof(c->aad), "%s %llu %u", c->source, (unsigned long long)c->sequence, c->batch);
+ }
if (c->body_size > PURE_OBJECT_LIMIT) { refuse(c, 413, "Content Too Large", "limit"); return; }
++host->requests;
Route route = {.kind = c->route};
@@ -1459,6 +1506,7 @@ static void accept_all(Host *host) {
c->out = NULL;
c->out_count = c->out_at = 0;
c->body = NULL;
+ c->opened = 0;
c->in_used = 0;
struct epoll_event event = {.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET, .data.ptr = c};
if (epoll_ctl(host->epoll, EPOLL_CTL_ADD, fd, &event)) {
@@ -1652,6 +1700,29 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen
return true;
}
+/* --seal OUTPUT=KEY: that output's sends sealed to the receiver's public key (hex). */
+static bool wire_seals(Executor *ex, const PureModule *m, const char *const *seals, unsigned count) {
+ for (unsigned i = 0; i < count; ++i) {
+ const char *equals = strchr(seals[i], '=');
+ unsigned char peer[PURE_SEAL_KEY];
+ char output[PURE_MODULE_NAME];
+ size_t length = equals ? (size_t)(equals - seals[i]) : 0;
+ PureWire *wire = NULL;
+ if (length && length < sizeof(output)) {
+ memcpy(output, seals[i], length);
+ output[length] = 0;
+ for (unsigned k = 0; k < ex->wire_count; ++k)
+ if (!strcmp(m->input_names[pure_wire_output(ex->wires[k])], output)) wire = ex->wires[k];
+ }
+ if (!wire || !pure_seal_peer(equals + 1, peer)) {
+ fprintf(stderr, "--seal takes OUTPUT=KEY for an output given to --send, the key as 64 hex digits: %s\n", seals[i]);
+ return false;
+ }
+ pure_wire_seal(wire, peer);
+ }
+ return true;
+}
+
/* --reply OUTPUT=INPUT: the answers to that output's sends become INPUT's events. */
static bool wire_replies(Executor *ex, const PureModule *m, const char *const *replies, unsigned count) {
for (unsigned i = 0; i < count; ++i) {
@@ -1690,14 +1761,18 @@ int main(int argc, char **argv) {
static PureModule module;
static Owner owner;
const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
- unsigned send_count = 0, reply_count = 0;
+ const char *seals[PURE_MODULE_INPUTS], *key_path = NULL;
+ unsigned send_count = 0, reply_count = 0, seal_count = 0;
unsigned port = 8767, threads = 1;
size_t budget = 64u << 20, capacity = 1024;
uint64_t timeout_ms = 10000;
for (int i = 1; i < argc; i += 2) {
uint64_t value;
+ if (!strcmp(argv[i], "--sealed-only")) { sealed_only = true; --i; continue; }
if (i + 1 == argc) goto usage;
if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
+ else if (!strcmp(argv[i], "--key")) key_path = argv[i + 1];
+ else if (!strcmp(argv[i], "--seal") && seal_count < PURE_MODULE_INPUTS) seals[seal_count++] = argv[i + 1];
else if (!strcmp(argv[i], "--store")) store = argv[i + 1];
else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1];
else if (!strcmp(argv[i], "--name")) name = argv[i + 1];
@@ -1715,6 +1790,12 @@ int main(int argc, char **argv) {
bool documents;
if (!module_path || !store || !load_module(&module, module_path, &documents, &show, &edit)) goto usage;
if (!threads) threads = auto_threads(module.states);
+ char public_key[2 * PURE_SEAL_KEY + 1] = "";
+ if (key_path) {
+ char error[256];
+ if (!(seal_key = pure_seal_key(key_path, public_key, error, sizeof(error)))) { fprintf(stderr, "--key %s\n", error); return 2; }
+ }
+ if (sealed_only && !seal_key) { fputs("--sealed-only needs --key\n", stderr); return 2; }
if (pure_store_init(store) != PURE_OK) { fputs("Cannot initialize the store directory\n", stderr); return 1; }
signal(SIGPIPE, SIG_IGN);
struct sigaction stop_action = {.sa_handler = stop};
@@ -1745,7 +1826,8 @@ int main(int argc, char **argv) {
owner.executor.reads_first = true;
owner.executor.publishes = true;
if (!open_wires(&owner.executor, &module, sends, send_count, name) ||
- !wire_replies(&owner.executor, &module, replies, reply_count)) return 2;
+ !wire_replies(&owner.executor, &module, replies, reply_count) ||
+ !wire_seals(&owner.executor, &module, seals, seal_count)) return 2;
after_save(&owner.executor); /* the snapshot, and the sends stored before */
for (unsigned i = 0; i < owner.executor.wire_count; ++i)
if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; }
@@ -1760,7 +1842,8 @@ int main(int argc, char **argv) {
if (!start_shard(&hosts[i], &module, documents, show, edit, store, budget, &port)) return 1;
}
if (owned && pthread_create(&owner_thread, NULL, run_owner, &owner)) { fputs("Cannot start the owner\n", stderr); return 1; }
- printf("{\"port\":%u,\"threads\":%u}\n", port, threads);
+ if (*public_key) printf("{\"port\":%u,\"threads\":%u,\"public_key\":\"%s\"}\n", port, threads, public_key);
+ else printf("{\"port\":%u,\"threads\":%u}\n", port, threads);
fflush(stdout);
for (unsigned i = 1; i < threads; ++i)
if (pthread_create(&thread[i], NULL, run_shard, &hosts[i])) { fputs("Cannot start a thread\n", stderr); return 1; }
@@ -1815,6 +1898,6 @@ int main(int argc, char **argv) {
usage:
fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
"[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] "
- "[--send-ca FILE] [--reply OUTPUT=INPUT ...]\n", stderr);
+ "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only]\n", stderr);
return 2;
}
diff --git a/tests/check.py b/tests/check.py
index 8352578..b80bf94 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -21,6 +21,9 @@ tests are core/test-module.c and store/test-tree-store.c.
batches to another app, which applies each once, across kill -9 of either
side; webhooks receive them over HTTPS only from a trusted receiver; an
API's answers become the app's events (--reply).
+7. Sealed sends: sealed to the receiver's key, they pass a relay unread and
+ unchanged (a changed byte is refused, a replay applies nothing), and a
+ node that keeps its deliveries sealed refuses unsealed ones.
"""
import argparse
import json
@@ -765,6 +768,137 @@ def server_links(aup, pure, work):
'attempts_after_restart': stats['a']['delivery_attempts']}
+def server_sealed(aup, pure, work):
+ """Sends sealed end to end (--seal to the receiver's --key): through a relay
+ that records what passes, every order is applied once and the relay saw
+ no item and no field name; one byte changed on the way is refused (400
+ seal) and applies nothing, the recorded request sent again applies
+ nothing twice, an unsealed delivery is refused (403) by a node that keeps
+ its deliveries sealed, and a seal to another key does not open."""
+ import select
+ import threading
+ import time
+ from http.client import HTTPConnection
+ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+ sender_module, receiver_module = work / 'seal-sender.pmod', work / 'seal-receiver.pmod'
+ for module, text in ((sender_module, LINK_SENDER), (receiver_module, LINK_RECEIVER)):
+ source = module.with_suffix('.aup')
+ source.write_text(text)
+ compile_module(aup, source, module)
+
+ def start(args):
+ """A host on a port of its own choosing (--port 0: no port probed
+ free and then taken by another socket before it binds)."""
+ process = subprocess.Popen([str(a) for a in args] + ['--port', '0'], stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE)
+ ready, _, _ = select.select([process.stdout], [], [], 10)
+ assert ready, 'host did not start'
+ line = process.stdout.readline()
+ assert line, process.communicate()[1][-1000:]
+ return process, json.loads(line)
+
+ def request(port, headers, body, path='/inputs/order'):
+ connection = HTTPConnection('127.0.0.1', port, timeout=10)
+ try:
+ connection.request('POST', path, body, headers)
+ response = connection.getresponse()
+ response.read()
+ return response.status, response.getheader('X-Pure-Error')
+ finally:
+ connection.close()
+
+ key = work / 'seal-receiver.key'
+ receiver, started = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'seal-b',
+ '--key', key, '--sealed-only'])
+ port_b, public_key = started['port'], started['public_key']
+ assert len(public_key) == 64 and (key.stat().st_mode & 0o777) == 0o600, (public_key, oct(key.stat().st_mode))
+ seen = []
+
+ class Relay(BaseHTTPRequestHandler):
+ protocol_version = 'HTTP/1.1'
+
+ def do_POST(self):
+ body = self.rfile.read(int(self.headers['Content-Length']))
+ headers = {name: value for name, value in self.headers.items()
+ if name.lower().startswith('x-pure-') or name.lower() in ('content-type', 'idempotency-key')}
+ seen.append((headers, body))
+ connection = HTTPConnection('127.0.0.1', port_b, timeout=10)
+ connection.request('POST', self.path, body, headers)
+ answer = connection.getresponse()
+ data = answer.read()
+ self.send_response(answer.status)
+ for name, value in answer.getheaders():
+ if name.lower().startswith('x-pure-'):
+ self.send_header(name, value)
+ self.send_header('Content-Length', str(len(data)))
+ self.end_headers()
+ self.wfile.write(data)
+ connection.close()
+
+ def log_message(self, *args):
+ pass
+
+ relay = ThreadingHTTPServer(('127.0.0.1', 0), Relay)
+ port_relay = relay.server_address[1]
+ threading.Thread(target=relay.serve_forever, daemon=True).start()
+ sender, sender_started = start([pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'seal-a',
+ '--send', f'ordered=http://127.0.0.1:{port_relay}/inputs/order',
+ '--seal', f'ordered={public_key}'])
+ port_a = sender_started['port']
+ stranger = None
+ try:
+ orders, total = 60, 0
+ for i in range(orders):
+ count = i % 7 + 1
+ total += count
+ assert request(port_a, {'Content-Type': 'application/json'},
+ json.dumps({'item': f'secret-{i}', 'count': count}).encode(), '/inputs/buy')[0] == 200
+ deadline = time.time() + 20
+ while int(http(port_b, 'GET', '/views/count')[3]) < orders and time.time() < deadline:
+ time.sleep(0.05)
+ assert int(http(port_b, 'GET', '/views/count')[3]) == orders
+ assert int(http(port_b, 'GET', '/views/total')[3]) == total
+ assert seen and all('X-Pure-Seal' in headers for headers, _ in seen), seen[:1]
+ assert not any(b'secret' in body or b'item' in body or b'city' in body for _, body in seen)
+ headers, body = seen[0]
+ changed = bytearray(body)
+ changed[len(changed) // 2] ^= 1
+ assert request(port_b, headers, bytes(changed)) == (400, 'seal')
+ assert request(port_b, headers, body)[0] == 200 # heard before: applied once
+ plain = {name: value for name, value in headers.items() if name != 'X-Pure-Seal'}
+ assert request(port_b, plain, json.dumps({'item': 'plain', 'count': 1, 'to': {'city': 'x'}}).encode()) == \
+ (403, 'unsealed')
+ # A sender sealing to another node's key: the receiver cannot open it.
+ other, other_started = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'seal-c',
+ '--key', work / 'seal-other.key'])
+ other.terminate()
+ other.wait()
+ stranger, stranger_started = start([pure / 'build/pure-serve', '--module', sender_module, '--store',
+ work / 'seal-d', '--send', f'ordered=http://127.0.0.1:{port_relay}/inputs/order',
+ '--seal', f'ordered={other_started["public_key"]}'])
+ before = len(seen)
+ assert request(stranger_started['port'], {'Content-Type': 'application/json'},
+ json.dumps({'item': 'stranger', 'count': 5}).encode(), '/inputs/buy')[0] == 200
+ deadline = time.time() + 5
+ while len(seen) == before and time.time() < deadline:
+ time.sleep(0.05)
+ time.sleep(0.2)
+ assert len(seen) > before and int(http(port_b, 'GET', '/views/count')[3]) == orders
+ # The same key file gives the same public key after a restart.
+ receiver.terminate()
+ receiver.wait()
+ receiver, again = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'seal-b',
+ '--key', key, '--sealed-only'])
+ assert again['public_key'] == public_key
+ return {'orders': orders, 'relayed': before, 'suite': seen[0][0]['X-Pure-Seal']}
+ finally:
+ for process in (sender, receiver, stranger):
+ if process:
+ process.terminate()
+ process.wait()
+ relay.shutdown()
+
+
def server_webhook(aup, pure, work):
"""An output wired to a third-party HTTPS endpoint (a Python server with
a certificate of its own): five orders arrive in order as JSON with the
@@ -1223,6 +1357,7 @@ def main():
report['server_timer_beats'] = server_timers(aup, pure, work)
report['server_links'] = server_links(aup, pure, work)
report['server_webhook'] = server_webhook(aup, pure, work)
+ report['server_sealed'] = server_sealed(aup, pure, work)
report['server_answers'] = server_answers(aup, pure, work)
print(json.dumps(report))
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 10:24:14 +0000