愛記システムの基本設計：DApps側である愛記システム　データの一元管理　アクセス制御

先までは、"愛記"についての記載で、どのようにブロックチェーンSNSに組み込んで実装していけばよいのか、概念的なところからアプローチ方法を記載していった。概念設計としてはひとまず終えた。次は、フェデレーションモデル全体の基本設計といえるところまで、基本設計書に着手できるようなところまで、概念を具体化していきたい。そして、それにつながるDApps側である「愛記システム」を、Pythonプログラムで開発していきたい。

愛の行動のPL,BSを決算書として、個人単位、市町村単位、で公表するような愛記システムというものを考えている。愛の行動のデータベースはブロックチェーンのプログラムであり、日々の愛の行動による愛貨の移動を決算書にまとめていきたい。なお、市町村のブロックチェーンのプログラムは以前にも記載している。その市町村のブロックチェーンのプログラムにつながる愛記システムを、DApps側であるPythonプログラムとして設計したい。その場合、基本設計をどのような手順で進めていけばよいか、詳しく見ていこう。

愛記システムを設計するための基本手順を以下に示す。このシステムは、Pythonを用いて市町村のブロックチェーンと連携し、個人および市町村単位での愛の行動のデータを収集、記録し、決算書（PL、BS）として公表するものである。

基本設計のステップ

要件定義
アーキテクチャ設計
データベース設計
API設計
ブロックチェーンインターフェース
決算書の生成
フロントエンド開発
テストとデプロイ

基本設計の各ステップを順番に進めることで、ブロックチェーンとDAppsとして繋がる「愛記システム」の詳細な設計が可能になる。各ステップでは、関係者との協議やレビューを通じて設計内容を確定していくことが重要である。

1．要件定義

まず、基本設計の最初のステップである要件定義をしていきたい。どのような機能が必要か、どのような問題を解決するのかを洗い出したい。要件定義はシステム設計の最初の重要なステップであり、システムが解決するべき問題と、必要な機能を明確に定義するプロセスである。以下に、愛記システムのプログラムに必要な機能と解決すべき問題を列挙してみよう。

機能要件

愛の行動の記録
愛貨の移動の記録
決算書の生成
個人および市町村単位でのデータの集約
データのブロックチェーンへの記録と取得
愛貨の管理
ユーザー管理
通知機能
レポート機能
ダッシュボード

非機能要件

セキュリティ
可用性
パフォーマンス
スケーラビリティ
ユーザビリティ
コンプライアンス

解決すべき問題

透明性と信頼性の確保
データの一元管理
愛の行動の促進
評価制度の確立
データのセキュリティとプライバシーの保護

これらの要件を基に、愛記システムの基本設計を進めていくことが重要である。次のステップでは、これらの要件を具体的なアーキテクチャ設計に反映していくことになる。まずは、要件定義の解決すべき問題を一つずつクリアにしていきたい。

データの一元管理

愛記システムにおけるデータの一元管理は、データの収集、統合、保存、アクセス制御、およびバックアップを適切に行うことが重要である。以下に、これを実現するための設計を示す。

1. データ収集

・タイミング:

ユーザーが愛の行動を記録したとき
愛貨の移動が発生したとき
定期的な集計およびレポート生成時

・データの種類:

愛の行動データ（行動の内容、日時、場所、関与者）
愛貨の移動データ（送信者、受信者、金額、移動日時）
決算データ（収支、資産、負債）
ユーザーデータ（ユーザーID、プロフィール、認証情報）
市町村データ（市町村ID、名前、地理情報）

・方法:

各種データを専用のAPIエンドポイントを通じて収集
フォーム入力、スキャン、GPSデータなどのユーザーインターフェースからの入力

2. データ統合

・タイミング:

リアルタイムでのデータ入力時
定期的なバッチ処理（例: 毎日夜間）

・データの種類:

各データソースから取得した生データを統合データベースに変換

・方法:

ETL（Extract, Transform, Load）プロセスを使用して、異なるフォーマットのデータを統合
リアルタイムデータのストリーム処理（例: Kafkaなどのデータストリーミングプラットフォーム）

3. データ保存

・タイミング:

データ収集時
データ統合後

・データの種類:

統合データベース（例: MongoDB、PostgreSQL）
ブロックチェーンデータ

・方法:

各データベースに適切にインデックスを設定し、効率的なデータ格納を実現
ブロックチェーンに重要データをハッシュ化して記録、ハッシュのみをオンチェーンに保存し、詳細データはオフチェーンに保存

4. アクセス制御

・タイミング:

データアクセスリクエスト時

・データの種類:

ユーザー認証情報
アクセス制御リスト（ACL）

・方法:

JWT（JSON Web Tokens）を使用してユーザー認証を行い、各APIリクエストの認証と認可を実施
ユーザーごとにアクセス権限を設定し、アクセス制御リストで管理
RBAC（Role-Based Access Control）を導入し、役割ごとにアクセス権限を定義

5. データのバックアップとリカバリ

・タイミング:

定期的なバックアップ（例: 毎日、毎週）
システム障害発生時

・データの種類:

全データベースのスナップショット
ブロックチェーンデータ

・方法:

定期的なバックアップスケジュールを設定し、バックアップをクラウドストレージやオフサイトストレージに保存
自動バックアップと手動バックアップの両方をサポート
データリカバリ計画を策定し、定期的にリカバリテストを実施

アクセス制御リスト（ACL）でのアクセス権限管理

各ユーザーまたは役割ごとにアクセス権限を設定し、ACLで管理する。具体的に、当方のフェデレーションモデルに当てはめると、愛貨の送信者・受信者、市町村のブロックチェーンのプログラム上での承認者、グローバルチェーンのプログラム上での承認者、DApps側のプログラムの利用者、市町村のブロックチェーンのプログラムの利用者、市町村のブロックチェーンのプログラムにも参加していない外部者、愛記システム運営管理者、共同システム開発者、新たなDApps側のプログラムを開発する各会社、そして総責任者、という人々がいる。いつ、だれに、どのように権限を与えるのか、具体的に見ていこう。

・役割とアクセス権限の定義

愛貨の送信者・受信者:
- アクセス権限: 自分の取引に関連するデータの読み取り・書き込み。
- エンドポイント: /transactions, /balance
市町村のブロックチェーンのプログラム上での承認者:
- アクセス権限: 自分の市町村の取引の承認・検証。
- エンドポイント: /municipal/approve
グローバルチェーンのプログラム上での承認者:
- アクセス権限: グローバルチェーンの取引の承認・検証。
- エンドポイント: /global/approve
DApps側のプログラムの利用者:
- アクセス権限: DAppsを利用する一般ユーザーとしてのアクセス。
- エンドポイント: /dapps/data
市町村のブロックチェーンのプログラムの利用者:
- アクセス権限: 市町村ブロックチェーンを利用する一般ユーザーとしてのアクセス。
- エンドポイント: /municipal/data
市町村のブロックチェーンのプログラムにも参加していない外部者:
- アクセス権限: 読み取り専用のアクセス。
- エンドポイント: /public/data
愛記システム運営管理者:
- アクセス権限: 全てのデータへの読み取り・書き込み・承認。
- エンドポイント: /admin
共同システム開発者:
- アクセス権限: システム開発に必要なデータへのアクセス。
- エンドポイント: /dev/data
新たなDApps側のプログラムを開発する各会社:
- アクセス権限: 開発に必要なデータへのアクセス。
- エンドポイント: /dev/data
総責任者:
- アクセス権限: 全てのデータへのフルアクセス。
- エンドポイント: * (全てのエンドポイント)

・アクセス制御の実装

1. ACLの設定

・acl.py:
# 役割ごとのアクセス権限を定義
ROLES_PERMISSIONS = {
"sender_receiver": ["/transactions", "/balance"],
"municipal_approver": ["/municipal/approve"],
"global_approver": ["/global/approve"],
"dapps_user": ["/dapps/data"],
"municipal_user": ["/municipal/data"],
"external_user": ["/public/data"],
"blockchain_admin": ["/admin"],
"system_developer": ["/dev/data"],
"dapps_developer": ["/dev/data"],
"chief_responsible": ["*"]
}

def has_access(role, endpoint):
permissions = ROLES_PERMISSIONS.get(role, [])
return "*" in permissions or endpoint in permissions

2. JWT認証とACLによるアクセス制御

・auth.py:
import jwt
from datetime import datetime, timedelta
from werkzeug.security import generate_password_hash, check_password_hash

SECRET_KEY = "your_secret_key"

# ユーザー情報に役割を含める
users = {
"user1": {"password": generate_password_hash("password1"), "role": "sender_receiver"},
"user2": {"password": generate_password_hash("password2"), "role": "municipal_approver"},
"user3": {"password": generate_password_hash("password3"), "role": "global_approver"},
"user4": {"password": generate_password_hash("password4"), "role": "dapps_user"},
"user5": {"password": generate_password_hash("password5"), "role": "municipal_user"},
"user6": {"password": generate_password_hash("password6"), "role": "external_user"},
"admin": {"password": generate_password_hash("adminpassword"), "role": "blockchain_admin"},
"dev_company": {"password": generate_password_hash("devpassword"), "role": "dapps_developer"},
"system_dev": {"password": generate_password_hash("systemdevpassword"), "role": "system_developer"},
"chief": {"password": generate_password_hash("chiefpassword"), "role": "chief_responsible"}
}

def authenticate(username, password):
user = users.get(username)
if user and check_password_hash(user["password"], password):
token = jwt.encode({
"username": username,
"role": user["role"],
"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
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None

3. Flaskアプリケーションでのアクセス制御

・app.py:
from flask import Flask, request, jsonify
from auth import authenticate, verify_token
from acl import has_access

app = Flask(__name__)

@app.route("/login", methods=["POST"])
def login():
data = request.json
username = data.get("username")
password = data.get("password")
token = authenticate(username, password)
if token:
return jsonify({"token": token}), 200
return jsonify({"message": "Invalid credentials"}), 401

@app.route("/transactions", methods=["POST"])
def transactions():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401

token_data = verify_token(token)
if not token_data:
return jsonify({"message": "Invalid token"}), 401

role = token_data["role"]

if not has_access(role, "/transactions"):
return jsonify({"message": "Access denied"}), 403

# 取引データの処理
return jsonify({"message": "Transaction data processed"}), 200

@app.route("/municipal/approve", methods=["POST"])
def municipal_approve():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401

token_data = verify_token(token)
if not token_data:
return jsonify({"message": "Invalid token"}), 401

role = token_data["role"]

if not has_access(role, "/municipal/approve"):
return jsonify({"message": "Access denied"}), 403

# 市町村承認データの処理
return jsonify({"message": "Municipal approval data processed"}), 200

@app.route("/global/approve", methods=["POST"])
def global_approve():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401

token_data = verify_token(token)
if not token_data:
return jsonify({"message": "Invalid token"}), 401

role = token_data["role"]

if not has_access(role, "/global/approve"):
return jsonify({"message": "Access denied"}), 403

# グローバル承認データの処理
return jsonify({"message": "Global approval data processed"}), 200

@app.route("/dapps/data", methods=["GET"])
def dapps_data():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401

token_data = verify_token(token)
if not token_data:
return jsonify({"message": "Invalid token"}), 401

role = token_data["role"]

if not has_access(role, "/dapps/data"):
return jsonify({"message": "Access denied"}), 403

# DAppsデータの処理
return jsonify({"message": "DApps data accessed"}), 200

@app.route("/municipal/data", methods=["GET"])
def municipal_data():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401

token_data = verify_token(token)
if not token_data:
return jsonify({"message": "Invalid token"}), 401

role = token_data["role"]

if not has_access(role, "/municipal/data"):
return jsonify({"message": "Access denied"}), 403

# 市町村データの処理
return jsonify({"message": "Municipal data accessed"}), 200

@app.route("/admin", methods=["GET", "POST"])
def admin():
token = request.headers.get("Authorization")
if not token:
return jsonify({"message": "Token is missing"}), 401

token_data = verify_token(token)
if not token_data:
return jsonify({"message": "Invalid token"}), 401

username = token_data["username"]
role = token_data["role"]

if not has_access(role, "/admin"):
return jsonify({"message": "Access denied"}), 403

# 管理者データの処理
return jsonify({"message": "Admin data accessed"}), 200

if __name__ == "__main__":
app.run(debug=True)

この設計により、各ユーザーまたは役割ごとに適切なアクセス権限を設定し、フェデレーションモデルにおけるセキュリティとアクセス制御を実現する。

いかがであろうか、今回はアクセス権限について記載した。ユーザーごとにアクセス権限を明確に定義する処理は、不正防止という意味でもとても重要性が高い。このように設計していくと良いのだろう。

続・ティール組織　研究会のブログ

ティール組織が話題になっているが、具現化するにはどうしたらよいか？
その研究を続けるにあたり、さらに次の形態である、続・ティール組織なるものまで視野に入れ、具体的な施策・行動内容を研究・支援する会。

愛記システムの基本設計：DApps側である愛記システム　データの一元管理　アクセス制御

基本設計のステップ

1．要件定義

機能要件

非機能要件

解決すべき問題

データの一元管理

アクセス制御リスト（ACL）でのアクセス権限管理

・アクセス制御の実装

1. ACLの設定

2. JWT認証とACLによるアクセス制御

3. Flaskアプリケーションでのアクセス制御

愛記システムの基本設計：DApps側である愛記システム データの一元管理 アクセス制御

基本設計のステップ

1．要件定義

機能要件

非機能要件

解決すべき問題

データの一元管理

アクセス制御リスト（ACL）でのアクセス権限管理

・アクセス制御の実装

1. ACLの設定

2. JWT認証とACLによるアクセス制御

3. Flaskアプリケーションでのアクセス制御

愛記システムの基本設計：DApps側である愛記システム　データの一元管理　アクセス制御