GovStack ビルディング ブロック: デジタル政府モジュールの実装
GovStack のモジュール型アプローチを使用してデジタル政府サービスを実装する方法: ビルディング ブロック ID、支払い、同意、デジタル記録、メッセージングのため。 GovSpecs 2.0 (2025-2027)、 スケーラブルで相互運用可能な PPE のパラダイムとして 20 か国以上で採用されています。
GovStack: デジタル・ガバメントのためのモジュール式フレームワーク
政府スタック は 2020 年に開始された国際的な取り組みです。 エストニア、ドイツ、 ITU (国際電気通信連合) e DIAL（デジタル・インパクト・アライアンス） と サービスを構築するために必要なツール、知識、ベストプラクティスを共有するという目標 毎回ゼロから始める必要がなく、デジタル視聴者を大規模に視聴できます。
基本的な考え方はシンプルですが強力です。国ごとに一枚岩のシステムを開発するのではなく、各国が 政府デジタル サービスは次のように分解されます。 ビルディングブロック (機能モジュール) それらは特定の機能 (ID、支払い、メッセージング、レジスターなど) を提供します。 それらを自由に組み合わせてサービスを構築できます。ビルディングブロックは、 相互運用可能、再利用可能、実装に依存しない: GovStack は、 仕様ソフトウェアではありません。
2025 年には、 ガヴスペック 2.0 (2025 ～ 2027 年の戦略)、GovStack は 新しいビルディングブロックを統合し、相互運用性仕様を更新し、定義することによってフレームワークを構築します。 デジタル容量が豊富な国をサポートする段階的な成熟度モデル 違う。 20 か国以上が GovStack アプローチを積極的に使用しています。
何を学ぶか
- GovStack の 9 つの基本的な構成要素とその技術仕様
- 既存のイタリアの PA サービス (SPID、CIE、pagoPA) を GovStack ビルディング ブロックにマッピングする方法
- GovStack リファレンス アーキテクチャ: 階層化モデルと統合バス
- GovSpecs 2.0: 2025 ～ 2027 年の戦略の新機能
- OpenID Connect を使用したビルディング ブロック ID の実践的な実装
- ビルディングブロック決済: 各国の決済システムとの統合
- 同意の構成要素: GDPR に準拠した同意管理
- GovStack がコンテキストに適しているかどうかを評価する方法
9 つの基本的な構成要素
GovStack は、以下をカバーする 9 つのコア ビルディング ブロック (新しい 2025 仕様で公開) を定義しています。 あらゆるデジタル政府サービスに必要な機能横断性:
| ビルディングブロック | 関数 | プロトコル/規格 | イタリア語の例 |
|---|---|---|---|
| 身元 | 認証とアイデンティティ管理 | OIDC、SAML、W3C DID | SPID、CIE、EUDI ウォレット |
| 支払い | 支払いと送金の処理 | ISO 20022、REST API | PayPA |
| 同意 | 同意の収集と管理 | GDPR、DPIA、OAuth スコープ | GDPR による設計による CMP |
| デジタルレジストリ | 権威ある登記簿の管理（登記所、土地登記簿） | REST API、FHIR、CKAN | ANPR、土地登記簿、専門登記簿 |
| メッセージング | 政府と国民の間の安全なコミュニケーション | SMTP、プッシュ、WebSocket、MQTT | IO アプリ、PEC、pagoPA 通知 |
| 情報仲介 | システム間の安全なデータ交換 | Xロード、REST、GraphQL | PDND、AgID の相互運用性 |
| 登録 | サービスへの個人/団体の登録 | REST API、OAuth 2.0 | INPS ポータル、SUAP、自治体ポータル |
| スケジューラ | 予定と予約の管理 | iCalendar、REST API | ヘルスカップ、デジタルカウンター |
| ワークフロー | プロセスと手順の調整 | BPMN、SAGAパターン、REST | PA練習管理システム |
GovStack リファレンス アーキテクチャ
GovStack は、ビルディング ブロックが層状に配置される層状アーキテクチャを提案しています。 明確な責任によって区別されます。
- レイヤ 0 - インフラストラクチャ: クラウド、ネットワーク、セキュリティ。 BB は特定のクラウドに依存しません。 AWS、Azure、GCP、またはオンプレミスのインフラストラクチャ上で実行できます。
- レイヤー 1 - コア構成要素: 9 つの基本的な BB。これらは自律型サービスであり、 標準化された RESTful API。各 BB には公開仕様 (OpenAPI + JSON スキーマ) があります。
- レイヤ 2 - 共有サービス: 集中ログ、サービス メッシュ、 APIゲートウェイ、サービスディスカバリ。これらのサービスはすべての BB をサポートします。
- レイヤ 3 - アプリケーション: 実際のデジタル サービス (学校への入学、 証明書の要求、印紙税の支払いなど）を基盤とする BB を調整します。
# Architettura di un servizio PA con Building Block GovStack
# Esempio: Servizio di iscrizione scolastica online
# Il servizio orchestra 5 Building Block:
# 1. Identity BB - autenticazione genitore con SPID/CIE
# 2. Registration BB - raccolta dati del bambino
# 3. Digital Registries BB - verifica iscrizione anagrafica
# 4. Consent BB - consenso al trattamento dati minori
# 5. Messaging BB - conferma iscrizione via IO App/email
# 6. Payments BB - pagamento tassa iscrizione via pagoPA
from dataclasses import dataclass
from typing import Optional
import httpx
@dataclass
class SchoolEnrollmentRequest:
parent_session_token: str # Token da Identity BB (SPID/CIE)
child_fiscal_code: str
school_code: str
year: int
class SchoolEnrollmentService:
"""
Servizio iscrizione scolastica che orchestra i Building Block GovStack.
Pattern: Saga orchestrator (gestione stati distribuiti con compensazioni).
"""
def __init__(
self,
identity_bb_url: str,
registry_bb_url: str,
consent_bb_url: str,
messaging_bb_url: str,
payments_bb_url: str
):
self.identity_url = identity_bb_url
self.registry_url = registry_bb_url
self.consent_url = consent_bb_url
self.messaging_url = messaging_bb_url
self.payments_url = payments_bb_url
async def process_enrollment(self, request: SchoolEnrollmentRequest) -> dict:
"""
Saga: processo iscrizione con compensazioni in caso di errore.
Ogni step è idempotente e reversibile.
"""
saga_log = []
try:
# Step 1: Verifica identità genitore tramite Identity BB
parent_identity = await self._verify_parent_identity(
request.parent_session_token
)
saga_log.append({"step": "identity_verified", "status": "ok"})
# Step 2: Verifica residenza bambino tramite Digital Registries BB
child_registry = await self._verify_child_in_registry(
request.child_fiscal_code,
parent_identity["fiscal_number"]
)
saga_log.append({"step": "registry_verified", "status": "ok"})
# Step 3: Raccolta consenso GDPR tramite Consent BB
consent_id = await self._collect_consent(
parent_identity["fiscal_number"],
purpose="school_enrollment_data_processing"
)
saga_log.append({"step": "consent_collected", "consent_id": consent_id})
# Step 4: Registrazione iscrizione tramite Registration BB
enrollment_id = await self._register_enrollment(
child_fc=request.child_fiscal_code,
school_code=request.school_code,
year=request.year,
parent_fc=parent_identity["fiscal_number"]
)
saga_log.append({"step": "enrollment_registered", "enrollment_id": enrollment_id})
# Step 5: Pagamento tassa (se prevista) tramite Payments BB
payment_url = await self._create_payment(
parent_fc=parent_identity["fiscal_number"],
enrollment_id=enrollment_id,
amount_cents=1500 # 15 euro
)
saga_log.append({"step": "payment_created", "payment_url": payment_url})
# Step 6: Notifica conferma tramite Messaging BB
await self._send_confirmation(
parent_fc=parent_identity["fiscal_number"],
enrollment_id=enrollment_id,
payment_url=payment_url
)
saga_log.append({"step": "notification_sent", "status": "ok"})
return {
"status": "success",
"enrollment_id": enrollment_id,
"payment_url": payment_url,
"saga_log": saga_log
}
except Exception as e:
# Compensazione: rollback degli step completati in ordine inverso
await self._compensate(saga_log, e)
raise
async def _verify_parent_identity(self, session_token: str) -> dict:
"""Chiama il Building Block Identity per validare il token SPID/CIE."""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.identity_url}/v1/tokens/validate",
json={"token": session_token}
)
response.raise_for_status()
return response.json() # Returns: fiscal_number, name, surname, etc.
async def _verify_child_in_registry(self, child_fc: str, parent_fc: str) -> dict:
"""Chiama il Building Block Digital Registries per verificare l'anagrafe."""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.registry_url}/v1/citizens/{child_fc}/family-relations",
params={"parent_fiscal_code": parent_fc}
)
response.raise_for_status()
return response.json()
async def _collect_consent(self, parent_fc: str, purpose: str) -> str:
"""Registra il consenso tramite il Building Block Consent."""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.consent_url}/v1/consents",
json={
"citizen_pseudonym": self._pseudonymize(parent_fc),
"purpose": purpose,
"legal_basis": "GDPR Art. 6(1)(e)",
"version": "2025-01"
}
)
response.raise_for_status()
return response.json()["consent_id"]
async def _compensate(self, saga_log: list, error: Exception):
"""
Compensazione Saga: rollback in ordine inverso degli step completati.
Garantisce consistenza anche in caso di errori parziali.
"""
for step in reversed(saga_log):
try:
if step["step"] == "consent_collected":
await self._revoke_consent(step["consent_id"])
elif step["step"] == "enrollment_registered":
await self._cancel_enrollment(step["enrollment_id"])
except Exception as compensation_error:
# Logga l'errore di compensazione ma continua
print(f"Compensation error for {step['step']}: {compensation_error}")
def _pseudonymize(self, fiscal_code: str) -> str:
import hashlib, hmac
key = b"secret-vault-key" # In produzione: usa un HSM
return hmac.new(key, fiscal_code.encode(), hashlib.sha256).hexdigest()[:32]GovSpecs 2.0: 2025 ～ 2027 年戦略の新機能
ガヴスペック 2.0、2025 年に GovStack によって発表され、比較された重要なアップデートが導入されています 以前のバージョンの仕様への変更:
- 4 レベルの成熟度モデル: 「初期段階」より (トレーニングとアーキテクチャのみ) 高レベル）から「高度な統合」（国家相互運用性フレームワークに統合された複数の BB）。 これにより、各国はそれぞれの優先順位と能力に応じて GovStack を導入できるようになります。
- 検証可能な資格情報のサポート: Building Block Identity には、次の仕様が含まれるようになりました。 eIDAS 2.0 およびヨーロッパの EUDI ウォレット プログラムと連携した W3C 検証可能な認証情報。
- BBへ: AI コンポーネント (言語モデル、 透明性と説明可能性に注意を払って、政府サービスにおける予測、分類）を行う。
- OpenAPI 3.1仕様: すべての BB API は OpenAPI 3.1 で文書化されるようになりました。 SwaggerUI を介した完全な JSON スキーマとテスト可能なサンプル。
- 適合性テスト: 実装を検証するための自動フレームワーク BB は GovStack 仕様に準拠しています。
OIDC を使用したビルディング ブロック ID の実装
Building Block Identity は最も重要であり、実装が最も複雑です。 GovStack は、次の API を公開する必要があると指定しています。
# Building Block Identity - Implementazione minima conforme GovStack
# OpenAPI 3.1 compatible - FastAPI implementation
from fastapi import FastAPI, HTTPException, Header, Depends
from fastapi.security import HTTPBearer
from pydantic import BaseModel
from typing import Optional
import jwt
app = FastAPI(
title="Identity Building Block",
description="GovStack Identity BB - Conforme a GovSpecs 2.0",
version="2.0.0"
)
security = HTTPBearer()
# --- Modelli ---
class TokenValidationRequest(BaseModel):
token: str
expected_acr: Optional[str] = None # Livello autenticazione richiesto
class IdentityResponse(BaseModel):
sub: str # Identificativo presso l'IdP
fiscal_number: Optional[str] = None # Codice Fiscale (SPID/CIE)
given_name: str
family_name: str
birthdate: Optional[str] = None
acr: str # Livello autenticazione effettivo
auth_time: int # Timestamp autenticazione
session_valid_until: int # Scadenza sessione
class SessionCreationRequest(BaseModel):
idp_id: str # Identity Provider scelto dall'utente
redirect_uri: str
acr_values: str = "https://www.spid.gov.it/SpidL2"
scope: list = ["openid", "profile"]
ui_locales: str = "it"
# --- Endpoints del Building Block Identity ---
@app.post("/v1/sessions",
summary="Avvia sessione di autenticazione",
tags=["Sessions"],
response_model=dict)
async def create_authentication_session(request: SessionCreationRequest) -> dict:
"""
GovStack Identity BB - Avvia il flusso di autenticazione.
Restituisce l'URL di redirect verso l'IdP.
"""
# Genera state e nonce per sicurezza
import secrets
session_id = secrets.token_urlsafe(32)
state = secrets.token_urlsafe(32)
nonce = secrets.token_urlsafe(32)
# Recupera metadata dell'IdP dalla federazione OIDC
idp_metadata = await get_idp_metadata(request.idp_id)
# Costruisce URL autorizzazione (PKCE + Request Object)
auth_url = build_oidc_auth_url(
idp_auth_endpoint=idp_metadata["authorization_endpoint"],
client_id=CLIENT_ID,
redirect_uri=request.redirect_uri,
scope=request.scope,
state=state,
nonce=nonce,
acr_values=request.acr_values,
private_key=SIGNING_KEY
)
# Salva sessione (Redis o DB)
await session_store.save(session_id, {
"state": state, "nonce": nonce, "idp_id": request.idp_id
})
return {"session_id": session_id, "authorization_url": auth_url}
@app.post("/v1/tokens/validate",
summary="Valida un token di sessione",
tags=["Tokens"],
response_model=IdentityResponse)
async def validate_token(request: TokenValidationRequest) -> IdentityResponse:
"""
GovStack Identity BB - Valida un token e restituisce l'identità.
I servizi chiamanti usano questo endpoint per verificare l'autenticazione.
"""
try:
# Decodifica e valida il token (firma, scadenza, audience)
claims = jwt.decode(
request.token,
JWKS,
algorithms=["RS256"],
audience=CLIENT_ID
)
# Verifica livello autenticazione se richiesto
if request.expected_acr:
actual_acr = claims.get("acr", "")
if not _meets_acr_requirement(actual_acr, request.expected_acr):
raise HTTPException(
status_code=403,
detail=f"Insufficient authentication level. Required: {request.expected_acr}"
)
return IdentityResponse(
sub=claims["sub"],
fiscal_number=claims.get("fiscal_number"),
given_name=claims["given_name"],
family_name=claims["family_name"],
birthdate=claims.get("birthdate"),
acr=claims.get("acr", ""),
auth_time=claims.get("auth_time", 0),
session_valid_until=claims.get("exp", 0)
)
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expired")
except jwt.InvalidTokenError as e:
raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
@app.delete("/v1/sessions/{session_id}",
summary="Termina sessione (logout)",
tags=["Sessions"])
async def end_session(session_id: str) -> dict:
"""
GovStack Identity BB - Logout con propagazione verso l'IdP.
Implementa OIDC Back-Channel Logout per notificare tutti i RP attivi.
"""
session = await session_store.get(session_id)
if not session:
raise HTTPException(status_code=404, detail="Session not found")
# Notifica logout all'IdP (OIDC Back-Channel Logout)
idp_metadata = await get_idp_metadata(session["idp_id"])
if "end_session_endpoint" in idp_metadata:
await propagate_logout(idp_metadata["end_session_endpoint"], session)
await session_store.delete(session_id)
return {"status": "session_terminated"}
def _meets_acr_requirement(actual: str, required: str) -> bool:
"""Verifica che il livello di autenticazione effettivo soddisfi il requisito."""
ACR_LEVELS = {
"https://www.spid.gov.it/SpidL1": 1,
"https://www.spid.gov.it/SpidL2": 2,
"https://www.spid.gov.it/SpidL3": 3,
"https://www.cie.gov.it/cie/aa": 2,
}
return ACR_LEVELS.get(actual, 0) >= ACR_LEVELS.get(required, 0)メッセージングの構成要素: IO アプリと政府通知
Building Block Messaging は、政府と国民の間のコミュニケーションを管理します。イタリアではこのサービスが GovStack Messaging 仕様により準拠したものは、 私アプリ (io.italia.it)、PagoPA S.p.A が管理する PA と通信するための全国的なアプリです。
# Building Block Messaging - Integrazione con IO App
# API REST di IO App per invio messaggi ai cittadini
import httpx
from pydantic import BaseModel
from typing import Optional
class IOMessage(BaseModel):
fiscal_code: str # Codice Fiscale del destinatario
time_to_live: int = 3600 # Secondi di validità notifica push
content: dict # Contenuto del messaggio
default_addresses: Optional[dict] = None # Fallback email
class IOAppClient:
"""
Client per le API di IO App.
Le API IO App sono disponibili su https://developer.io.italia.it
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.io.italia.it/api/v1"
async def send_message(self, message: IOMessage) -> dict:
"""
Invia un messaggio a un cittadino tramite IO App.
Il cittadino deve aver attivato il proprio profilo IO.
"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/messages",
headers={
"Ocp-Apim-Subscription-Key": self.api_key,
"Content-Type": "application/json"
},
json={
"fiscal_code": message.fiscal_code,
"time_to_live": message.time_to_live,
"content": message.content,
"default_addresses": message.default_addresses
}
)
response.raise_for_status()
return response.json()
async def check_profile(self, fiscal_code: str) -> bool:
"""
Verifica se un cittadino ha attivato il profilo IO e accetta messaggi dalla PA.
"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/profiles/{fiscal_code}",
headers={"Ocp-Apim-Subscription-Key": self.api_key}
)
if response.status_code == 404:
return False
response.raise_for_status()
profile = response.json()
return profile.get("sender_allowed", False)
# Utilizzo: notifica iscrizione scolastica
async def notify_enrollment(fiscal_code: str, enrollment_id: str, payment_url: str):
io_client = IOAppClient(api_key="your-io-api-key")
# Verifica se il cittadino usa IO App
has_io = await io_client.check_profile(fiscal_code)
if has_io:
# Messaggio strutturato IO App con CTA pagamento
message = IOMessage(
fiscal_code=fiscal_code,
content={
"subject": f"Iscrizione Scolastica {enrollment_id} - Conferma",
"markdown": f"""
## La tua iscrizione è stata registrata
L'iscrizione con codice **{enrollment_id}** è stata registrata con successo.
Per completare la procedura, effettua il pagamento della tassa di iscrizione tramite il
link qui sotto.
**Importo**: 15,00 €
[Paga ora]({payment_url})
Per assistenza: [Ufficio Scolastico](https://istruzione.comune.esempio.it)
""",
"payment_data": {
"amount": 1500,
"notice_number": enrollment_id,
"payee": {
"fiscal_code": "COMUNE_FC",
"name": "Comune di Esempio"
}
}
},
default_addresses={"email": None} # No fallback email
)
await io_client.send_message(message)
else:
# Fallback: email tradizionale (da implementare)
await send_email_notification(fiscal_code, enrollment_id, payment_url)イタリアのエコシステム上の GovStack マップ
イタリアにはすでに先進的な PPE エコシステムがあり、GovStack ビルディング ブロックに自然にマッピングされています。 イタリアの開発者にとって、GovStack の採用はゼロから始めることを意味するのではなく、 標準化されたモジュールモデルに従って既存の統合を再編成する.
| BB ガバメントスタック | イタリアの実装 | 成熟度レベル | 埋めるギャップ |
|---|---|---|---|
| 身元 | SPID、CIE、eIDAS | 高 (レベル 3 ～ 4) | 完全な OIDC 統合、EUDI ウォレット |
| 支払い | PayPA | 高（レベル4） | 完全な GPD REST API |
| メッセージング | IO アプリ、PEC | 高（レベル3） | IO アプリの普及率 (一部の地域ではまだ低い) |
| デジタルレジストリ | ANPR、土地登記簿、PDND | 中 (レベル 2 ～ 3) | レジスタ間の相互運用性はまだ部分的 |
| 情報仲介 | PDND | 中（レベル2） | PDNDの採用は依然として拡大中 |
| 同意 | いろいろ（規格化されていない） | 低 (レベル 1) | 一元化されたコンセンサスプラットフォームが必要です |
| スケジューラ | 地域健康CUP | 低～中 (レベル 1～2) | 地域の細分化。国家基準がない |
| ワークフロー | PA 練習システム (異種混合) | 低 (レベル 1) | 大きな異質性。標準化が必要です |
GovStack を採用する時期
GovStack は、次の場合に特に適しています。
- あなたが構築しているのは、 新しいPAデジタルサービス ゼロから開発し、ベンダーへの依存を避けたい
- 既存のシステムを統合する必要があり、共有の参照アーキテクチャ モデルが必要である
- コンテキスト内で操作する 複数の国 （国際協力、国境を越えたサービス）
- GovStack オープンソース エコシステムに貢献し、他国の実装から恩恵を受けたいと考えている
イタリアのコンテキストに非常に特殊なサービスがある場合、GovStack はあまり適していません。 既存のもの (SPID、pagoPA) は、追加の抽象化のオーバーヘッドなしですでに要件を満たしています。 または、分散フェデレーションの複雑さを管理するためのリソースが限られている場合。
結論: GovTech シリーズの展望
この記事で、行政のデジタル化に特化した GovTech シリーズを終了します。 公共デジタル インフラストラクチャ (DPI) からアイデンティティに至るまで、完全なエコシステムを調査しました 欧州連合 (eIDAS 2.0、EUDI ウォレット)、SPID と CIE の具体的な実装からデータ保護まで (GDPR-by-Design)、アクセシビリティ (WCAG 2.1 AA) からオープン データ (DCAT-AP_IT、CKAN) まで 国際モジュラーフレームワーク (GovStack)。
共通点は常に同じです。公共デジタル サービスは次のとおりである必要があります。 包括的、 安全、相互運用可能、再利用可能。標準化されたビルディング ブロックを備えた GovStack は、次のことを実現します。 これらの目標を地球規模で達成するための共通の語彙と共有のアーキテクチャ モデル。
GovTech シリーズ全体
- #00: デジタル公共インフラ - アーキテクチャとビルディング ブロック
- #01: eIDAS 2.0 および EUDI ウォレット - 開発者ガイド
- #02: 政府アイデンティティのための OpenID Connect
- #03: オープン データ API 設計 - パブリック データの公開と利用
- #04: GDPR-by-Design - 公共サービスのアーキテクチャ パターン
- #05: PA - WCAG 2.1 AA 実装用のアクセス可能な UI
- #06: 政府 API 統合 - SPID、CIE、IT デジタル サービス
- #07: GovStack ビルディング ブロック - (この記事)