← Back to Research
Overview
Blockchain-backed delivery proof and tamper-proof audit trails for email compliance and legal evidence.
Problem Statement
Current email audit logs: - Mutable: Can be altered by administrators - Not provable: No cryptographic proof of delivery - Centralized trust: Must trust the MTA - No timestamp proof: Timestamps can be backdated - Limited legal standing: "My server says..." is weak evidence
Vision
Immutable Audit Trail
Every email event (sent, delivered, read, deleted) is:
✅ Cryptographically signed
✅ Timestamped by trusted authority
✅ Anchored to blockchain (optional)
✅ Independently verifiable
✅ Tamper-proof
✅ Legal-grade evidence
Use Cases
1. Legal Compliance
- Prove message was delivered at specific time
- Non-repudiable delivery receipts
- Regulatory compliance (SOX, HIPAA, GDPR)
- E-discovery requests
2. Dispute Resolution
- Contract signing via email
- Proof of notice
- Timestamp verification
- Chain of custody
3. Security Audits
- Detect unauthorized access
- Prove no tampering
- Compliance audits
- Incident response
Architecture
1. Merkle Tree Audit Log
Root Hash (anchored to blockchain)
/ \
Hash(A,B) Hash(C,D)
/ \ / \
Event A Event B Event C Event D
Each event = {
type: "message_sent",
timestamp: "2026-03-07T20:00:00Z",
message_id: "abc123@msgs.global",
from: "alice@msgs.global",
to: "bob@msgs.global",
hash: sha256(event_data),
prev_hash: hash_of_previous_event
}
2. Trusted Timestamp Service
msgs.global → [Event] → Timestamp Authority → [RFC 3161 Timestamp Token]
|
Blockchain Anchor (optional)
|
↓ Verifiable Proof ↓
3. Blockchain Anchoring
class BlockchainAnchor:
"""Anchor audit log root hash to blockchain"""
def __init__(self, blockchain='ethereum'):
self.blockchain = blockchain
self.contract_address = '0x...'
def anchor_merkle_root(self, root_hash, batch_size=1000):
"""
Anchor Merkle root to blockchain
Instead of one transaction per event (expensive),
batch 1000 events into Merkle tree, anchor root
"""
# Build Merkle tree from batch of events
events = get_pending_events(batch_size)
merkle_tree = build_merkle_tree(events)
root_hash = merkle_tree.root
# Anchor root to blockchain
tx_hash = self.write_to_blockchain(root_hash)
# Store blockchain reference
db.execute("""
INSERT INTO blockchain_anchors (
merkle_root, blockchain, tx_hash, event_count, anchored_at
) VALUES (?, ?, ?, ?, NOW())
""", root_hash, self.blockchain, tx_hash, len(events))
# Update events with blockchain proof
for event in events:
proof = merkle_tree.get_proof(event)
db.execute("""
UPDATE audit_events
SET merkle_proof = ?, blockchain_tx = ?
WHERE id = ?
""", json.dumps(proof), tx_hash, event.id)
return tx_hash
def verify_event(self, event_id):
"""
Verify event was included in blockchain-anchored batch
Returns proof that anyone can independently verify
"""
event = db.query("SELECT * FROM audit_events WHERE id = ?", event_id)
# Get blockchain transaction
tx = get_blockchain_transaction(event.blockchain_tx)
# Verify Merkle proof
is_valid = verify_merkle_proof(
event.event_hash,
event.merkle_proof,
tx.data # Root hash stored in blockchain
)
return {
'valid': is_valid,
'blockchain': event.blockchain,
'tx_hash': event.blockchain_tx,
'timestamp': tx.timestamp,
'block_number': tx.block_number,
'confirmations': get_confirmations(tx.block_number)
}
Database Schema
-- Immutable audit events
CREATE TABLE audit_events (
id SERIAL PRIMARY KEY,
event_type VARCHAR(50) NOT NULL, -- message_sent, delivered, read, deleted
event_timestamp TIMESTAMP NOT NULL,
message_id VARCHAR(255),
user_id INTEGER REFERENCES users(id),
event_data JSONB NOT NULL,
event_hash VARCHAR(64) NOT NULL, -- SHA-256 of event
prev_hash VARCHAR(64), -- Hash chain
signature VARCHAR(256), -- Cryptographic signature
merkle_proof JSONB, -- Proof for blockchain verification
blockchain_tx VARCHAR(66), -- Ethereum tx hash
created_at TIMESTAMP DEFAULT NOW()
);
-- Blockchain anchors (batch roots)
CREATE TABLE blockchain_anchors (
id SERIAL PRIMARY KEY,
merkle_root VARCHAR(64) UNIQUE NOT NULL,
blockchain VARCHAR(20), -- ethereum, bitcoin, etc.
tx_hash VARCHAR(66) UNIQUE NOT NULL,
block_number BIGINT,
event_count INTEGER,
anchored_at TIMESTAMP DEFAULT NOW(),
confirmations INTEGER
);
-- Timestamp tokens (RFC 3161)
CREATE TABLE timestamp_tokens (
id SERIAL PRIMARY KEY,
event_id INTEGER REFERENCES audit_events(id),
tsa_url VARCHAR(255), -- Timestamp authority URL
token BYTEA, -- ASN.1 encoded timestamp token
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_audit_events_message ON audit_events(message_id);
CREATE INDEX idx_audit_events_timestamp ON audit_events(event_timestamp);
CREATE INDEX idx_audit_events_blockchain ON audit_events(blockchain_tx);
Implementation
1. Event Logging
class AuditLogger:
"""Tamper-proof audit logging"""
def log_event(self, event_type, message_id, user_id, event_data):
"""Log event to immutable audit trail"""
# Get previous event hash (for chaining)
prev_event = db.query("""
SELECT event_hash FROM audit_events
ORDER BY id DESC LIMIT 1
""")
prev_hash = prev_event.event_hash if prev_event else '0' * 64
# Create event
event = {
'event_type': event_type,
'event_timestamp': datetime.now().isoformat(),
'message_id': message_id,
'user_id': user_id,
'event_data': event_data,
'prev_hash': prev_hash
}
# Hash event
event_hash = hashlib.sha256(
json.dumps(event, sort_keys=True).encode()
).hexdigest()
event['event_hash'] = event_hash
# Sign event
signature = self.sign_event(event_hash)
# Store
db.execute("""
INSERT INTO audit_events (
event_type, event_timestamp, message_id, user_id,
event_data, event_hash, prev_hash, signature
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", event_type, event['event_timestamp'], message_id, user_id,
json.dumps(event_data), event_hash, prev_hash, signature)
# Get trusted timestamp (RFC 3161)
self.get_timestamp_token(event_hash)
return event_hash
def sign_event(self, event_hash):
"""Sign event with server private key"""
# Load server private key
with open('/etc/msgs.global/audit-key.pem') as f:
private_key = load_pem_private_key(f.read())
# Sign
signature = private_key.sign(
event_hash.encode(),
padding.PSS(mgf=padding.MGF1(hashes.SHA256()), ...)
)
return base64.b64encode(signature).decode()
def get_timestamp_token(self, event_hash):
"""Get RFC 3161 timestamp token from TSA"""
# Create timestamp request
tsr = TimeStampReq()
tsr.setMessageImprint(hashlib.sha256(event_hash.encode()).digest())
# Send to timestamp authority
response = requests.post(
'https://freetsa.org/tsr', # Free TSA (or use commercial)
data=tsr.to_der(),
headers={'Content-Type': 'application/timestamp-query'}
)
# Parse timestamp token
tst = TimeStampToken(response.content)
# Store token
db.execute("""
INSERT INTO timestamp_tokens (event_id, tsa_url, token)
VALUES (?, ?, ?)
""", event.id, 'https://freetsa.org/tsr', response.content)
return tst
2. Message Lifecycle Events
# Hook into message processing
@app.after_send_message
def log_message_sent(message):
"""Log message sent event"""
audit_logger.log_event(
'message_sent',
message.message_id,
message.from_user_id,
{
'to': message.to,
'subject': message.subject,
'size': len(message.body),
'attachments': len(message.attachments)
}
)
@app.after_message_delivered
def log_message_delivered(message, recipient):
"""Log delivery event"""
audit_logger.log_event(
'message_delivered',
message.message_id,
recipient.user_id,
{
'delivered_to': recipient.email,
'delivery_time': datetime.now().isoformat(),
'mta': recipient.receiving_mta
}
)
@app.after_message_read
def log_message_read(message, user):
"""Log read event (IMAP SEEN flag)"""
audit_logger.log_event(
'message_read',
message.message_id,
user.id,
{
'read_at': datetime.now().isoformat(),
'client': request.headers.get('User-Agent')
}
)
@app.after_message_deleted
def log_message_deleted(message, user):
"""Log deletion event"""
audit_logger.log_event(
'message_deleted',
message.message_id,
user.id,
{
'deleted_at': datetime.now().isoformat(),
'reason': 'user_request'
}
)
3. Verification API
@app.route('/api/v1/audit/verify/<event_id>')
def verify_audit_event(event_id):
"""Verify audit event integrity"""
event = db.query("SELECT * FROM audit_events WHERE id = ?", event_id)
# 1. Verify hash chain
chain_valid = verify_hash_chain(event)
# 2. Verify signature
signature_valid = verify_event_signature(event)
# 3. Verify timestamp token
timestamp_valid = verify_timestamp_token(event)
# 4. Verify blockchain anchor (if exists)
blockchain_valid = None
if event.blockchain_tx:
blockchain_valid = verify_blockchain_anchor(event)
return jsonify({
'event_id': event_id,
'event_type': event.event_type,
'timestamp': event.event_timestamp,
'verification': {
'hash_chain': chain_valid,
'signature': signature_valid,
'timestamp': timestamp_valid,
'blockchain': blockchain_valid
},
'trusted_timestamp': get_trusted_timestamp(event),
'blockchain_confirmations': get_blockchain_confirmations(event) if event.blockchain_tx else None
})
@app.route('/api/v1/audit/proof/<message_id>')
def get_delivery_proof(message_id):
"""Get cryptographic proof of delivery"""
events = db.query("""
SELECT * FROM audit_events
WHERE message_id = ?
ORDER BY event_timestamp
""", message_id)
proof = {
'message_id': message_id,
'sent_event': find_event(events, 'message_sent'),
'delivered_event': find_event(events, 'message_delivered'),
'read_event': find_event(events, 'message_read'),
'blockchain_anchor': get_blockchain_anchor(events),
'timestamp_tokens': get_timestamp_tokens(events),
'verification_url': f'https://msgs.global/verify/{message_id}'
}
return jsonify(proof)
4. Blockchain Anchoring (Background Job)
#!/usr/bin/env python3
"""Periodic blockchain anchoring"""
from web3 import Web3
from eth_account import Account
def anchor_to_blockchain():
"""Anchor audit logs to Ethereum every hour"""
# Get unanchored events (last hour)
events = db.query("""
SELECT * FROM audit_events
WHERE blockchain_tx IS NULL
AND created_at > NOW() - INTERVAL '1 hour'
""")
if not events:
return
# Build Merkle tree
merkle_tree = MerkleTree([e.event_hash for e in events])
root_hash = merkle_tree.root
# Connect to Ethereum
w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/YOUR_KEY'))
account = Account.from_key(os.getenv('ETH_PRIVATE_KEY'))
# Smart contract call (simple data storage)
contract = w3.eth.contract(address=CONTRACT_ADDRESS, abi=CONTRACT_ABI)
tx = contract.functions.storeHash(root_hash).buildTransaction({
'from': account.address,
'nonce': w3.eth.get_transaction_count(account.address),
'gas': 100000,
'gasPrice': w3.eth.gas_price
})
# Sign and send
signed_tx = account.sign_transaction(tx)
tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)
print(f"Anchored {len(events)} events to Ethereum: {tx_hash.hex()}")
# Update database
db.execute("""
INSERT INTO blockchain_anchors (
merkle_root, blockchain, tx_hash, event_count
) VALUES (?, 'ethereum', ?, ?)
""", root_hash, tx_hash.hex(), len(events))
# Update events with proof
for event in events:
proof = merkle_tree.get_proof(event.event_hash)
db.execute("""
UPDATE audit_events
SET merkle_proof = ?, blockchain_tx = ?
WHERE id = ?
""", json.dumps(proof), tx_hash.hex(), event.id)
if __name__ == '__main__':
anchor_to_blockchain()
Cost Analysis
Blockchain Anchoring Costs
Option 1: Ethereum Mainnet - Per transaction: ~$2-10 (varies with gas) - Batch size: 1,000 events - Cost per event: $0.002-0.01 - Frequency: Hourly batches
Option 2: Polygon (L2) - Per transaction: ~$0.01 - Batch size: 1,000 events - Cost per event: $0.00001 - Frequency: Every 15 minutes
Option 3: Timestamps.io (Free) - Free Bitcoin timestamping - No Ethereum costs - Less flexible than custom contract
Advantages
- Legally defensible: Cryptographic proof in court
- Non-repudiable: Cannot deny sending/receiving
- Tamper-proof: Blockchain immutability
- Independently verifiable: Anyone can verify
- Compliance: Meets strictest audit requirements
Status
📋 Research & Design Phase
Next Steps
- [ ] Implement hash-chained audit log
- [ ] Integrate RFC 3161 timestamp service
- [ ] Build Merkle tree batching
- [ ] Deploy Ethereum smart contract for anchoring
- [ ] Create verification UI
- [ ] Legal review of proof validity
- [ ] Performance testing (write amplification)
Related Standards
- RFC 3161: Time-Stamp Protocol (TSP)
- RFC 6962: Certificate Transparency (Merkle tree usage)
- ISO 18014: Time-Stamping Services
- NIST SP 800-102: Recommendation for Digital Signature Timeliness