مستندات فنی CipherMQ

نمای کلی و معماری

هسته سیستم حول یک مدل آشنا برای هرکسی که با RabbitMQ کار کرده باشد می‌چرخد: exchange → binding (routing key) → queue. اما برخلاف RabbitMQ، CipherMQ از پروتکل استاندارد AMQP استفاده نمی‌کند؛ یک پروتکل متنی خطی و اختصاصی روی سوکت TLS خام پیاده‌سازی شده است (جزئیات در بخش پروتکل سیمی). تفاوت مهم دیگر این است که بروکر هرگز محتوای پیام را نمی‌بیند: پیام‌ها پیش از رسیدن به سرور توسط کلاینت رمزنگاری می‌شوند و سرور صرفاً یک رله‌ی رمزنگاری‌کور (blind relay) است.

هویت هر کلاینت و بر همین اساس نقش، مجوزها، و سطل‌های محدودسازی نرخ او مستقیماً از Common Name (CN) گواهی TLS ارائه‌شده در زمان دست‌دهی mTLS استخراج می‌شود. هیچ لایه جداگانه‌ی نام‌کاربری/گذرواژه یا توکنی در مسیر داده وجود ندارد.

مدل امنیتی و احراز هویت

۲.۱ احراز هویت متقابل (mTLS)

هر دو listener (بروکر اصلی و کنسول مدیریتی) با rustls::ServerConfig و WebPkiClientVerifier پیکربندی می‌شوند تا گواهی کلاینت را در برابر یک RootCertStore ساخته‌شده از ca_cert_path اعتبارسنجی کنند. اتصالی که گواهی معتبر امضاشده توسط همان CA ارائه ندهد، در همان مرحله دست‌دهی TLS رد می‌شود — پیش از آنکه حتی یک بایت داده‌ی برنامه رد و بدل شود.

پس از دست‌دهی موفق، CN از فیلد Subject گواهی (OID 2.5.4.3) با x509-parser استخراج می‌شود و به‌عنوان client_id در سراسر سیستم استفاده می‌شود.

۲.۲ کنترل دسترسی مبتنی بر نقش (ACL)

AclManager هر CN را به یکی از چهار نقش نگاشت می‌کند: Sender، Receiver، Admin یا Unknown (بر اساس عضویت در سه فهرست پیکربندی‌شده acl.sender_cns، acl.receiver_cns، acl.admin_cns). نقش Unknown از همان ابتدا و پیش از رسیدن به حلقه دستورات رد می‌شود.

علاوه‌بر بررسی سطح دستور، برای declare_queue و bind یک بررسی سطح منبع (check_resource) نیز اجرا می‌شود: یک Receiver فقط می‌تواند صفی را اعلام یا bind کند که نامش با "{cn}_" شروع شود یا دقیقاً "{cn}_queue" باشد؛ routing key نیز باید با CN شروع شود. در مقابل، declare_exchange هیچ بررسی مالکیتی ندارد هر Receiver می‌تواند هر نام exchange دلخواهی را اعلام کند.

۲.۳ سرور مدیریتی

admin_server.rs پس از دست‌دهی TLS، بلافاصله CN را استخراج و نقش را حل می‌کند؛ اگر نقش دقیقاً Admin نباشد، اتصال بی‌صدا بسته می‌شود (بدون ارسال پیام خطا به کلاینت). پارامتر پیکربندی admin.allowed_cns نیز به تابع serve پاس داده می‌شود اما در امضای تابع با _allowed_cns نام‌گذاری شده و در منطق واقعی مجوزدهی استفاده نمی‌شود (جزئیات در بخش محدودیت‌ها).

رمزنگاری: در انتقال، سر تا سر، و در سکون

۳.۱ رمزنگاری در انتقال (TLS)

تمام ترافیک چه بروکر اصلی و چه کنسول مدیریتی از طریق TLS 1.2/1.3 با پیاده‌سازی rustls عبور می‌کند و mTLS برای هر دو طرف اجباری است. گواهی سرور و کلید خصوصی آن با rustls_pemfile بارگذاری می‌شوند؛ کلید خصوصی باید در قالب PKCS8 باشد (pkcs8_private_keys) کلیدهای PKCS1/RSA خام (BEGIN RSA PRIVATE KEY) شناسایی نمی‌شوند و باید پیش از استفاده تبدیل شوند.

۳.۲ رمزنگاری سر تا سر پیام (End-to-End)

بروکر به‌طور طراحی‌شده محتوای پیام را نمی‌بیند. ساختار EncryptedInputData که ناشران ارسال می‌کنند فقط شامل فیلدهای زیر است:

pub struct EncryptedInputData {
    pub message_id: String,
    pub receiver_client_id: String,
    pub enc_session_key: String,  // کلید نشست، رمزشده برای گیرنده
    pub nonce: String,
    pub ciphertext: String,       // بدنه واقعی پیام، هرگز plaintext نیست
}

وجود enc_session_key در کنار nonce و ciphertext نشان‌دهنده یک الگوی envelope encryption در سمت کلاینت است: یک کلید نشست متقارن یک‌بارمصرف، بدنه پیام را رمز می‌کند و خودِ آن کلید نشست با کلید عمومی گیرنده رمز می‌شود. بروکر نه کلید نشست را می‌تواند بگشاید و نه بدنه پیام را صرفاً بایت‌ها را ذخیره (در حافظه) و منتقل می‌کند.

برای این مدل، بروکر یک دایرکتوری ساده از کلید عمومی ارائه می‌دهد: register_public_key (فقط Receiver، همیشه روی client_id خودِ فراخوان) و get_public_key (فقط Sender، بدون محدودیت روی این‌که کلید عمومی چه کسی خوانده شود که با توجه به ماهیت «عمومی» بودن این کلید، منطقی است).

۳.۳ رمزنگاری در سکون (At Rest)

کلیدهای عمومی ثبت‌شده پیش از نوشتن در PostgreSQL با AES-256-GCM (کتابخانه aes_gcm) رمز می‌شوند: یک nonce تصادفی ۹۶ بیتی برای هر نوشتن تولید می‌شود، و ciphertext/tag/nonce به‌صورت جداگانه و base64-encoded در جدول public_keys ذخیره می‌شوند. کلید رمزنگاری از encryption.aes_key در فایل پیکربندی (base64، دقیقاً ۳۲ بایت پس از دیکد) خوانده می‌شود و برای کل سرور مشترک است نه به‌ازای هر کلاینت یا هر tenant.

توجه: بدنه پیام‌ها (ciphertext پیام) هرگز در دیتابیس ذخیره نمی‌شود فقط متادیتای پیام (شناسه، فرستنده، exchange، routing key، زمان‌ها) پایدار می‌گردد. جزئیات و پیامدهای این موضوع در بخش تضمین‌های تحویل شرح داده شده است.

پروتکل سیمی و مرجع دستورات

برخلاف RabbitMQ که از AMQP 0-9-1 استفاده می‌کند، CipherMQ یک پروتکل متنی خطی اختصاصی روی سوکت TLS خام پیاده‌سازی کرده است. هر درخواست یک رشته UTF-8 به‌شکل COMMAND arg1 arg2 ... argN است که با جداکننده‌ی فاصله تفکیک می‌شود؛ استخراج آرگومان‌ها معمولاً با splitn انجام می‌شود، به‌این معنا که فقط آخرین آرگومان (مثلاً بدنه JSON در publish) می‌تواند خودش شامل فاصله باشد نام exchange، صف، یا routing key نباید فاصله داشته باشند.

۴.۱ دستورات بروکر اصلی

۴.۲ دستورات کنسول مدیریتی

چرخه حیات پیام

۵.۱ انتشار (Publish)

1. بررسی message_status.contains_key(message_id)؛ در صورت وجود، پیام به‌عنوان تکراری رد می‌شود (idempotency در سطح شناسه پیام).
2. ساخت رکورد MessageMetadata با sent_time (UTC RFC3339) و نوشتن همزمان (synchronous) آن در PostgreSQL از طریق storage.save_metadata این نوشتن پیش از پاسخ ACK به ناشر تکمیل می‌شود.
3. یافتن صف‌های مقصد: بررسی bindings[exchange] و فیلتر بر اساس تطابق دقیق routing_key (بدون الگوی wildcard مانند # یا * در AMQP).
4. اگر هیچ صفی matched نشود، خطا بازگردانده می‌شود با این‌حال رکورد متادیتا از قدم قبل همچنان در دیتابیس باقی می‌ماند.
5. برای هر صف مقصد: در صورت رسیدن به max_queue_size، قدیمی‌ترین پیام آن صف حذف (evict) می‌شود؛ سپس پیام جدید در انتهای VecDeque صف قرار می‌گیرد.
6. مصرف‌کنندگان زنده صف (از طریق کانال mpsc::UnboundedSender) مطلع می‌شوند؛ فرستنده‌های بسته‌شده در همین گذر پاک‌سازی می‌شوند.
7. ACK نهایی (ACK <message_id>) به ناشر بازگردانده می‌شود.

sequenceDiagram
    participant Pub as Publisher
    participant Srv as handle_client
    participant Acl as AclManager
    participant Rl as RateLimiter
    participant St as ServerState
    participant Db as Storage worker

    Pub->>Srv: publish exch key json
    Srv->>Acl: check_command(role, publish)
    Acl-->>Srv: Allowed
    Srv->>Rl: check(cn, ip, is_publish=true)
    Rl-->>Srv: Allowed
    Srv->>St: publish(exch, key, message)
    St->>St: dedupe by message_id
    St->>Db: SaveMetadata (sync, awaits reply)
    Db-->>St: Ok
    St->>St: resolve bound queues by routing key
    St->>St: enqueue into each queue (VecDeque)
    St->>St: notify live consumers via mpsc
    St-->>Srv: Ok
    Srv-->>Pub: ACK message_id

نمودار ۲ مسیر publish. نوشتن متادیتا (نه بدنه پیام) پیش از ACK به ناشر تضمین می‌شود.

۵.۲ مصرف و تایید (Consume & Ack)

دستور consume صف را (در صورت نبود) اعلام می‌کند، فرستنده کانال کلاینت را به فهرست مصرف‌کنندگان آن صف اضافه می‌کند، و بدون ارسال هیچ پاسخ فوری به کلاینت پس از ۱۰۰ میلی‌ثانیه تاخیر (برای پرهیز از race با پاسخ ثبت مصرف‌کننده)، پیام‌های در انتظارِ قابل‌تحویل صف را با فاصله ۱۰ میلی‌ثانیه بین هر پیام برای مصرف‌کننده جدید بازپخش می‌کند.

sequenceDiagram
    participant Con as Consumer
    participant Srv as handle_client
    participant St as ServerState
    participant Db as Storage worker

    Con->>Srv: consume queue_name
    Srv->>St: register_consumer(queue, sender)
    St->>St: declare_queue if missing
    St->>St: push sender into consumers map
    Note over Con,St: no immediate response sent
    St->>St: after 100ms, replay deliverable pending messages
    St-->>Con: "Message: id json" via mpsc -> socket
    Note over Con,Srv: consumer processes message
    Con->>Srv: ack message_id
    Srv->>St: acknowledge(message_id)
    St->>St: remove from EVERY bound queue + tracking maps
    St->>Db: UpdateAcknowledgedTimeAsync (buffered)
    Srv-->>Con: ACK confirmed message_id

نمودار ۳ مسیر consume/ack. توجه شود که ack، پیام را از تمام صف‌های مقصدش حذف می‌کند، نه فقط از صفی که این مصرف‌کننده روی آن ثبت شده.

تضمین‌های تحویل پیام

با جمع‌بندی رفتارهای بخش‌های قبل، تضمین تحویل CipherMQ را می‌توان این‌گونه خلاصه کرد:

مدیریت وضعیت و همزمانی

ServerState به‌طور کامل بدون یک قفل سراسری واحد ساخته شده است؛ هر نگاشت (صف‌ها، binding‌ها، exchange‌ها، مصرف‌کنندگان، وضعیت پیام‌ها) یک DashMap مجزاست که قفل را در سطح shard نگه می‌دارد، و شمارنده اتصال فعال یک AtomicUsize است نه RwLock<usize>.

stateDiagram-v2
    [*] --> Sent: publish() persists metadata
    Sent --> Delivered: pushed to a consumer socket
    Delivered --> Acknowledged: client sends ack
    Delivered --> Expired: unacked past cleanup window (default ~50min)
    Sent --> Evicted: queue reached max_queue_size
    Acknowledged --> [*]
    Expired --> [*]
    Evicted --> [*]

نمودار ۴ وضعیت‌های ممکن یک پیام از دید message_status.

لایه ذخیره‌سازی

Storage از یک الگوی actor استفاده می‌کند: یک تسک پس‌زمینه‌ی واحد، مالکیت انحصاری PgPool را در اختیار دارد و تمام عملیات دیتابیس از طریق کانال mpsc::channel::<StorageCommand> به آن ارسال می‌شوند نه از طریق دسترسی مستقیم و همزمان به pool از چندین تسک.

• Pool: حداکثر ۲۰۰ / حداقل ۱۰ اتصال، طول عمر حداکثر ۱ ساعت، idle timeout ۱۰۰ دقیقه، acquire timeout ۵۰ دقیقه.
• اتصال اولیه: تا ۵ تلاش با فاصله ۵ ثانیه‌ای؛ در صورت شکست همه، راه‌اندازی سرور با خطا متوقف می‌شود.
• بازیابی حین اجرا: اگر pool.is_closed() شود، تسک storage پیش از پردازش دستور بعدی، یک تلاش اتصال مجدد انجام می‌دهد.
• دسته‌بندی نوشتن (batching): به‌روزرسانی‌های delivered_time و acknowledged_time در یک HashMap بافر می‌شوند و هر ۵۰۰ میلی‌ثانیه یا در رسیدن به ۵۰ آیتم، با یک کوئری دسته‌ای (UPDATE ... FROM unnest($1::text[])) نوشته می‌شوند برخلاف نوشتن اولیه متادیتا (SaveMetadata) که همزمان و بلافاصله انجام می‌شود، چون ACK به ناشر به آن وابسته است.

محدودسازی نرخ و محافظت در برابر اضافه‌بار

الگوریتم پایه token bucket کلاسیک است: در هر take()، تعداد توکن‌ها بر اساس زمان سپری‌شده از آخرین شارژ و نرخ پیکربندی‌شده (rps) تا سقف burst بازپر می‌شود؛ اگر حداقل یک توکن موجود باشد، مصرف و درخواست پذیرفته می‌شود.

تاب‌آوری و مدیریت خطا

همچنین، در main.rs، هنگام شکست دست‌دهی TLS، state.decrement_client_count() فراخوانی می‌شود درحالی‌که increment_client_count() فقط پس از عبور موفق از بررسی نقش و rate limiter در handle_client اجرا می‌شود. این عدم تقارن به‌معنای کاهش شمارنده بدون افزایش متناظر است.

سرور مدیریتی، متریک‌ها و لاگ

کنسول مدیریتی (پیش‌فرض 127.0.0.1:9091، غیرفعال مگر admin.enabled = true) یک پروتکل متنی مشابه بروکر اصلی ولی جداگانه ارائه می‌دهد و صرفاً برای نقش Admin در دسترس است.

۱۱.۱ متریک‌ها

خروجی metrics در قالب سبک Prometheus (# HELP / # TYPE) است:

دستور json یک تصویر کامل‌تر ارائه می‌دهد که شامل هم شمارنده‌های اتمیک زنده (live) و هم آخرین ServerSnapshot ذخیره‌شده است این دو ممکن است به دلیل ماهیت به‌روزرسانی دوره‌ای snapshot، لحظه‌ای با هم یکسان نباشند.

۱۱.۲ لاگ‌گیری

بر پایه tracing + tracing-subscriber: سه فایل جداگانه و چرخشی (rolling::hourly/daily/never بر اساس logging.rotation) برای سطوح info/debug/error به‌صورت JSON، به‌علاوه یک لایه‌ی «pretty» روی stdout که با RUST_LOG یا logging.level فیلتر می‌شود.

پیکربندی (config.toml)

پیکربندی با toml + serde بارگذاری و اعتبارسنجی می‌شود؛ نمونه‌ای معادل بخش‌های تعریف‌شده در config.rs:

# اتصال اصلی بروکر
[server]
address = "0.0.0.0:5671"
connection_type = "tls"          # فعلاً تنها مقدار پشتیبانی‌شده

[tls]
cert_path = "/etc/ciphermq/certs/server.crt"
key_path = "/etc/ciphermq/certs/server.key"   # باید PKCS8 باشد
ca_cert_path = "/etc/ciphermq/certs/ca.crt"

[logging]
level = "info"
rotation = "daily"               # hourly | daily | هر مقدار دیگر = never
info_file_path = "logs/info.log"
debug_file_path = "logs/debug.log"
error_file_path = "logs/error.log"
max_size_mb = 50

[database]
host = "127.0.0.1"
port = 5432
user = "ciphermq"
password = "********"
dbname = "ciphermq"

[encryption]
algorithm = "x25519_chacha20_poly1305"   # فقط اعتبارسنجی می‌شود، رجوع به بخش رمزنگاری
aes_key = "<base64 دقیقاً معادل ۳۲ بایت>"

[performance]
max_queue_size = 10000
heartbeat_timeout = 60
cleanup_interval = 3000
consumer_cleanup_interval = 60
idle_timeout_sec = 60

[admin]
enabled = true
address = "127.0.0.1:9091"
allowed_cns = []                 # بارگذاری می‌شود اما در مجوزدهی واقعی استفاده نمی‌شود

[rate_limit]
enabled = true
global_rps = 100.0
global_burst = 200.0
publish_rps = 50.0
publish_burst = 100.0
max_connections_per_ip = 50
max_connections_per_cn = 20

[acl]
sender_cns = ["publisher-svc-1"]
receiver_cns = ["worker-svc-1"]
admin_cns = ["ops-admin"]

قوانین اعتبارسنجی اجباری در Config::load:

• اگر connection_type = "tls"، هر سه مسیر tls.* الزامی‌اند.
• database.host، database.user، database.dbname نباید خالی باشند.
• encryption.algorithm باید دقیقاً "x25519_chacha20_poly1305" باشد.
• encryption.aes_key باید base64 معتبر و پس از دیکد دقیقاً ۳۲ بایت باشد.