先までは、"愛記"についての記載で、どのようにブロックチェーンSNSに組み込んで実装していけばよいのか、概念的なところからアプローチ方法を記載していった。概念設計としてはひとまず終えた。次は、フェデレーションモデル全体の基本設計といえるところまで、基本設計書に着手できるようなところまで、概念を具体化していきたい。そして、それにつながるDApps側である「愛記システム」を、Pythonプログラムで開発していきたい。
愛の行動のPL,BSを決算書として、個人単位、市町村単位、で公表するような愛記システムというものを考えている。愛の行動のデータベースはブロックチェーンのプログラムであり、日々の愛の行動による愛貨の移動を決算書にまとめていきたい。なお、市町村のブロックチェーンのプログラムは以前にも記載している。その市町村のブロックチェーンのプログラムにつながる愛記システムを、DApps側であるPythonプログラムとして設計したい。その場合、基本設計をどのような手順で進めていけばよいか、詳しく見ていこう。
愛記システムを設計するための基本手順を以下に示す。このシステムは、Pythonを用いて市町村のブロックチェーンと連携し、個人および市町村単位での愛の行動のデータを収集、記録し、決算書（PL、BS）として公表するものである。
基本設計のステップ
- 要件定義
- アーキテクチャ設計
- データベース設計
- API設計
- ブロックチェーンインターフェース
- 決算書の生成
- フロントエンド開発
- テストとデプロイ
基本設計の各ステップを順番に進めることで、ブロックチェーンとDAppsとして繋がる「愛記システム」の詳細な設計が可能になる。各ステップでは、関係者との協議やレビューを通じて設計内容を確定していくことが重要である。
1．要件定義
まず、基本設計の最初のステップである要件定義をしていきたい。どのような機能が必要か、どのような問題を解決するのかを洗い出したい。要件定義はシステム設計の最初の重要なステップであり、システムが解決するべき問題と、必要な機能を明確に定義するプロセスである。以下に、愛記システムのプログラムに必要な機能と解決すべき問題を列挙してみよう。
機能要件
-
愛の行動の記録
-
愛貨の移動の記録
-
決算書の生成
-
個人および市町村単位でのデータの集約
-
データのブロックチェーンへの記録と取得
-
愛貨の管理
-
ユーザー管理
-
通知機能
-
レポート機能
-
ダッシュボード
非機能要件
-
セキュリティ
-
可用性
-
パフォーマンス
-
スケーラビリティ
-
ユーザビリティ
-
コンプライアンス
解決すべき問題
-
透明性と信頼性の確保
-
データの一元管理
-
愛の行動の促進
-
評価制度の確立
-
データのセキュリティとプライバシーの保護
これらの要件を基に、愛記システムの基本設計を進めていくことが重要である。次のステップでは、これらの要件を具体的なアーキテクチャ設計に反映していくことになる。まずは、要件定義の解決すべき問題を一つずつクリアにしていきたい。
透明性と信頼性の確保
1. データの完全性と一貫性の保証
-
データの完全性: すべての取引と愛の行動記録が正確かつ改ざんされていないことを保証する。
-
データの一貫性: システム全体でデータが統一されていることを確認する。
2. 取引の透明性の確保
-
公開取引データ: 取引データを公開し、誰でも確認できるようにする。
-
監査ログ: 取引や行動のすべての変更履歴を保持し、監査可能にする。
3. データのセキュリティとプライバシー保護
-
暗号化: データの送受信時および保存時に暗号化を行い、データの安全性を確保。
-
アクセス制御: データへのアクセスを制御し、必要な権限を持つユーザーのみに限定。
4. 取引の正当性の確認
-
署名の生成と検証: 取引データの署名を生成し、取引の正当性を検証。
-
取引の検証: 各取引を検証し、不正行為や二重支出を防止。
これらの項目を詳細に決定し、実装することで、愛記システムの透明性と信頼性を確保することができる。各項目については、具体的な技術要件や設計仕様を定義し、システム開発の各フェーズで反映させることが重要である。
取引の検証について
取引の正当性の確認において、特に不正行為や二重支出を防止するための設計は、取引の検証を行うための一連のステップを設けることが重要である。以下に、具体的な設計とそのプログラム実装例を示す。
-
取引の検証タイミング
- タイミング: 取引がDAppsに到達した時点で、取引の検証を行う。
- 内容: 取引データの一貫性と正当性を確認し、不正行為や二重支出を防止。
-
取引の検証項目
- 署名の検証: 取引データに含まれる署名を検証し、送信者が正当な所有者であることを確認。
- 二重支出の防止: 取引が二重に使用されていないかを確認。
- 残高確認: 送信者の残高が取引額に対して十分であることを確認。
- タイムスタンプの確認: 取引が適切な時間内に行われたことを確認。
-
設計手順
- 取引データの受信。
- 署名の検証。
- 二重支出の確認。
- 残高の確認。
- タイムスタンプの確認。
- 取引の承認または拒否。
以下に、Pythonでの取引検証の実装例を示す。この例では、Flaskを使用してDAppsのエンドポイントを作成し、取引の検証を行う。
import hashlib
import jwt
from datetime import datetime, timedelta
from flask import Flask, request, jsonify
import requests
import random
app = Flask(__name__)
SECRET_KEY = "your_secret_key"
# データベースの模擬（実際には適切なデータベースを使用する）
users = {
"user1": {"password": "password1", "balance": 100.0, "public_key": [random.getrandbits(8) for _ in range(64)]},
"user2": {"password": "password2", "balance": 150.0, "public_key": [random.getrandbits(8) for _ in range(64)]},
}
transactions = []
def generate_ntru_keys():
public_key = [random.getrandbits(8) for _ in range(64)]
private_key = [random.getrandbits(8) for _ in range(64)]
return public_key, private_key
def ntru_encrypt(data, public_key):
return [d ^ k for d, k in zip(data, public_key)]
def ntru_sign(data, private_key):
return [d ^ k for d, k in zip(data, private_key)]
def ntru_decrypt(encrypted_data, private_key):
return [e ^ k for e, k in zip(encrypted_data, private_key)]
def sign_transaction(data, private_key):
transaction_hash = hashlib.sha256(data.encode()).digest()
signature = ntru_sign(transaction_hash, private_key)
return signature
def verify_signature(data, signature, public_key):
transaction_hash = hashlib.sha256(data.encode()).digest()
calculated_signature = ntru_sign(transaction_hash, public_key)
return signature == calculated_signature
def authenticate(username, password):
if username in users and users[username]["password"] == password:
token = jwt.encode({
"username": username,
"exp": datetime.utcnow() + timedelta(hours=1)
}, SECRET_KEY, algorithm="HS256")
return token
return None
def verify_token(token):
try:
data = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
return data["username"]
except jwt.ExpiredSignatureError:
return None
def is_double_spending(transaction_id):
return any(tx["transaction_id"] == transaction_id for tx in transactions)
def has_sufficient_balance(username, amount):
return users[username]["balance"] >= amount
@app.route("/transactions", methods=["POST"])
def receive_transaction():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401
username = verify_token(token)
if not username:
return jsonify({"message": "Invalid token"}), 401
transaction_data = request.json.get("transaction_data")
signature = request.json.get("signature")
public_key = users[username]["public_key"]
transaction_id = request.json.get("transaction_id")
amount = request.json.get("amount")
timestamp = request.json.get("timestamp")
if verify_signature(transaction_data, signature, public_key):
if is_double_spending(transaction_id):
return jsonify({"message": "Double spending detected"}), 400
if not has_sufficient_balance(username, amount):
return jsonify({"message": "Insufficient balance"}), 400
try:
txn_time = datetime.fromisoformat(timestamp)
if abs((datetime.utcnow() - txn_time).total_seconds()) > 300: # 5分以内の取引のみ有効
return jsonify({"message": "Invalid timestamp"}), 400
except ValueError:
return jsonify({"message": "Invalid timestamp format"}), 400
# 取引の保存
transactions.append({
"transaction_id": transaction_id,
"username": username,
"amount": amount,
"timestamp": datetime.utcnow().isoformat(),
"signature": signature
})
users[username]["balance"] -= amount
# 市町村のブロックチェーンに送信
response = requests.post("http://localhost:8081/transactions", json={
"transaction_id": transaction_id,
"username": username,
"amount": amount,
"timestamp": datetime.utcnow().isoformat(),
"signature": signature
})
if response.status_code == 200:
return jsonify({"message": "Transaction is valid and sent to municipal chain"}), 200
else:
return jsonify({"message": "Failed to send transaction to municipal chain"}), 500
else:
return jsonify({"message": "Invalid transaction"}), 400
if __name__ == "__main__":
app.run(debug=True)
・市町村のブロックチェーンのRustコード
市町村のブロックチェーンのプログラムを、ラティス署名とラティス暗号を使用して、DApps側から送信されたトランザクションを受信して処理するように更新する。
use actix_web::{web, App, HttpServer, HttpResponse, Responder};
use serde::{Serialize, Deserialize};
use sha2::{Sha256, Digest};
use chrono::{DateTime, Utc};
use std::sync::Mutex;
use std::collections::HashMap;
use rand::Rng;
#[derive(Serialize, Deserialize)]
struct Transaction {
transaction_id: String,
username: String,
amount: f64,
timestamp: String,
signature: Vec<u8>,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct NTRUKeys {
pub public_key: Vec<u8>,
pub private_key: Vec<u8>,
}
pub fn generate_ntru_keys() -> NTRUKeys {
let mut rng = rand::thread_rng();
let public_key = (0..64).map(|_| rng.gen::<u8>()).collect();
let private_key = (0..64).map(|_| rng.gen::<u8>()).collect();
NTRUKeys { public_key, private_key }
}
pub fn ntru_encrypt(data: &[u8], public_key: &[u8]) -> Vec<u8> {
data.iter().zip(public_key).map(|(&d, &k)| d ^ k).collect()
}
pub fn ntru_sign(data: &[u8], private_key: &[u8]) -> Vec<u8> {
private_key.iter().zip(data).map(|(&k, &d)| k ^ d).collect()
}
pub fn ntru_decrypt(encrypted_data: &[u8], private_key: &[u8]) -> Vec<u8> {
encrypted_data.iter().zip(private_key).map(|(&e, &k)| e ^ k).collect()
}
pub fn sign_transaction(data: &[u8], private_key: &[u8]) -> Vec<u8> {
private_key.iter().zip(data).map(|(&k, &d)| k ^ d).collect()
}
pub fn verify_signature(data: &[u8], signature: &[u8], public_key: &[u8]) -> bool {
let transaction_hash = Sha256::digest(data);
let calculated_signature = ntru_sign(&transaction_hash, public_key);
calculated_signature == signature
}
struct AppState {
users: Mutex<HashMap<String, NTRUKeys>>,
transactions: Mutex<Vec<Transaction>>,
jwt_secret: String,
}
async fn receive_transaction(data: web::Data<AppState>, transaction: web::Json<Transaction>) -> impl Responder {
let transaction_data = format!(
"{}{}{}{}",
transaction.transaction_id, transaction.username, transaction.amount, transaction.timestamp
);
let transaction_data_bytes = transaction_data.as_bytes();
let users = data.users.lock().unwrap();
if let Some(user_keys) = users.get(&transaction.username) {
if verify_signature(transaction_data_bytes, &transaction.signature, &user_keys.public_key) {
let mut transactions = data.transactions.lock().unwrap();
if transactions.iter().any(|tx| tx.transaction_id == transaction.transaction_id) {
return HttpResponse::BadRequest().body("Double spending detected");
}
transactions.push(transaction.into_inner());
return HttpResponse::Ok().body("Transaction is valid");
} else {
return HttpResponse::BadRequest().body("Invalid signature");
}
} else {
return HttpResponse::BadRequest().body("User not found");
}
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
let mut users = HashMap::new();
let user1_keys = generate_ntru_keys();
users.insert("user1".to_string(), user1_keys);
let app_state = web::Data::new(AppState {
users: Mutex::new(users),
transactions: Mutex::new(Vec::new()),
jwt_secret: "your_secret_key".to_string(),
});
HttpServer::new(move || {
App::new()
.app_data(app_state.clone())
.route("/transactions", web::post().to(receive_transaction))
})
.bind("0.0.0.0:8081")?
.run()
.await
}
・説明
-
データ構造と暗号化関数の定義:
- NTRUKeys構造体を定義し、generate_ntru_keys、ntru_encrypt、ntru_sign、ntru_decrypt関数を実装した。
-
署名検証:
- verify_signature関数で、受信したトランザクションデータの署名を検証する。データをSHA256でハッシュ化し、NTRU署名で署名を検証する。
-
取引データの受信エンドポイント:
- /transactionsエンドポイントで、受信したトランザクションを検証し、有効であれば保存する。
- 二重支出の確認も行い、取引IDが既に存在する場合はエラーメッセージを返す。
Pythonコードの requests.post 関数を使用して、市町村のブロックチェーンの /transactions エンドポイントにトランザクションデータを送信している。これにより、DAppsと市町村のブロックチェーン間の連携が実現されている。以上の設計と実装により、DApps側のPythonコードと市町村のブロックチェーンプログラムの連携がラティス署名とラティス暗号を使用して実現されている。
いかがであろうか、今回は取引データの検証について記載した。DApps側だけでなく、各プロセスにおいて検証は必要であろうから、フェデレーションモデル全体にかかわる話でもあるのだろう。もっと具体的に各プロセスにおいてどうしていくのかの設計も必要だろう。