Python项目--基于区块链的门生证书认证体系

缠丝猫 · 3 天前

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要登录才可以下载或查看，没有账号？立即注册

x

引言

在当今数字化时代，学历造假和证书伪造已成为教育领域的一大挑战。传统的证书验证方法每每耗时且容易被篡改，这促使我们寻求更加安全、透明和高效的解决方案。区块链技能以其不可篡改、去中心化和透明的特性，为门生证书认证提供了抱负的技能底子。
本文将详细介绍怎样利用Python和区块链技能构建一个完备的门生证书认证体系，该体系答应教育机构安全地颁发数字证书，同时使门生和雇主可以大概轻松验证这些证书的真实性。
体系概述

我们的基于区块链的门生证书认证体系重要包含以下几个焦点组件：

区块链网络：用于存储证书哈希和元数据
证书天生器：为门生创建数字证书
证书验证器：验证证书的真实性
Web界面：用户友爱的前端界面
数据库：存储用户信息和证书详情

体系工作流程如下：

教育机构通过体系为门生天生数字证书
证书的哈希值被记录在区块链上
门生可以分享其数字证书
雇主或其他相关方可以通过体系验证证书的真实性

技能栈

本项目利用以下技能栈：

Python 3.9+：焦点编程语言
Flask：Web应用框架
Ethereum：区块链平台
Web3.py：与以太坊区块链交互的Python库
SQLAlchemy：ORM工具
PyPDF2：PDF处理惩罚
cryptography：加密功能
HTML/CSS/JavaScript：前端开辟

体系计划与实现

1. 项目布局

certificate_system/
├── app/
│ ├── __init__.py
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ ├── certificate.py
│ │ └── institution.py
│ ├── blockchain/
│ │ ├── __init__.py
│ │ ├── contract.py
│ │ └── ethereum.py
│ ├── certificate/
│ │ ├── __init__.py
│ │ ├── generator.py
│ │ └── validator.py
│ ├── routes/
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ ├── certificate.py
│ │ └── institution.py
│ ├── static/
│ │ ├── css/
│ │ ├── js/
│ │ └── images/
│ └── templates/
├── config.py
├── requirements.txt
└── run.py

复制代码

2. 数据库模型计划

我们利用SQLAlchemy计划了以下焦点数据模型：
用户模型 (User)

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(128))
role = db.Column(db.String(20), default='student')
created_at = db.Column(db.DateTime, default=datetime.utcnow)
certificates = db.relationship('Certificate', backref='owner', lazy='dynamic')
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)

复制代码

教育机构模型 (Institution)

class Institution(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
description = db.Column(db.Text)
website = db.Column(db.String(200))
ethereum_address = db.Column(db.String(42), unique=True, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
certificates = db.relationship('Certificate', backref='institution', lazy='dynamic')
admin_id = db.Column(db.Integer, db.ForeignKey('user.id'))

复制代码

证书模型 (Certificate)

class Certificate(db.Model):
id = db.Column(db.Integer, primary_key=True)
certificate_id = db.Column(db.String(64), unique=True, nullable=False)
title = db.Column(db.String(100), nullable=False)
description = db.Column(db.Text)
issue_date = db.Column(db.DateTime, default=datetime.utcnow)
expiry_date = db.Column(db.DateTime, nullable=True)
hash_value = db.Column(db.String(66), unique=True, nullable=False)
blockchain_tx = db.Column(db.String(66), unique=True)
pdf_path = db.Column(db.String(200))
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
institution_id = db.Column(db.Integer, db.ForeignKey('institution.id'))
is_revoked = db.Column(db.Boolean, default=False)

复制代码

3. 区块链集成

我们利用以太坊区块链和智能合约来存储证书哈希。以下是智能合约的简化版本：

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;
contract CertificateRegistry {
struct Certificate {
bytes32 hashValue;
address issuer;
uint256 timestamp;
bool isRevoked;
}
mapping(bytes32 => Certificate) public certificates;
event CertificateIssued(bytes32 indexed certificateId, bytes32 hashValue, address indexed issuer);
event CertificateRevoked(bytes32 indexed certificateId);
function issueCertificate(bytes32 certificateId, bytes32 hashValue) public {
require(certificates[certificateId].timestamp == 0, "Certificate already exists");
certificates[certificateId] = Certificate({
hashValue: hashValue,
issuer: msg.sender,
timestamp: block.timestamp,
isRevoked: false
});
emit CertificateIssued(certificateId, hashValue, msg.sender);
}
function revokeCertificate(bytes32 certificateId) public {
require(certificates[certificateId].timestamp > 0, "Certificate does not exist");
require(certificates[certificateId].issuer == msg.sender, "Only issuer can revoke");
require(!certificates[certificateId].isRevoked, "Certificate already revoked");
certificates[certificateId].isRevoked = true;
emit CertificateRevoked(certificateId);
}
function verifyCertificate(bytes32 certificateId, bytes32 hashValue) public view returns (bool, address, uint256, bool) {
Certificate memory cert = certificates[certificateId];
if (cert.timestamp == 0) {
return (false, address(0), 0, false);
}
bool isValid = (cert.hashValue == hashValue) && !cert.isRevoked;
return (isValid, cert.issuer, cert.timestamp, cert.isRevoked);
}
}

复制代码

在Python中，我们利用Web3.py库与智能合约交互：

from web3 import Web3
import json
class EthereumClient:
def __init__(self, provider_url, contract_address, abi_path, private_key=None):
self.web3 = Web3(Web3.HTTPProvider(provider_url))
with open(abi_path, 'r') as f:
contract_abi = json.load(f)
self.contract = self.web3.eth.contract(address=contract_address, abi=contract_abi)
self.private_key = private_key
def issue_certificate(self, certificate_id, hash_value, issuer_address):
certificate_id_bytes = Web3.toBytes(hexstr=certificate_id)
hash_value_bytes = Web3.toBytes(hexstr=hash_value)
nonce = self.web3.eth.getTransactionCount(issuer_address)
txn = self.contract.functions.issueCertificate(
certificate_id_bytes,
hash_value_bytes
).buildTransaction({
'chainId': 1,
'gas': 200000,
'gasPrice': self.web3.toWei('50', 'gwei'),
'nonce': nonce,
})
signed_txn = self.web3.eth.account.signTransaction(txn, private_key=self.private_key)
tx_hash = self.web3.eth.sendRawTransaction(signed_txn.rawTransaction)
return self.web3.toHex(tx_hash)
def verify_certificate(self, certificate_id, hash_value):
certificate_id_bytes = Web3.toBytes(hexstr=certificate_id)
hash_value_bytes = Web3.toBytes(hexstr=hash_value)
result = self.contract.functions.verifyCertificate(
certificate_id_bytes,
hash_value_bytes
).call()
return {
'is_valid': result[0],
'issuer': result[1],
'timestamp': result[2],
'is_revoked': result[3]
}

复制代码

4. 证书天生与验证

证书天生

import hashlib
import uuid
from datetime import datetime
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
from reportlab.lib import colors
from reportlab.lib.styles import getSampleStyleSheet
from reportlab.platypus import Paragraph, Table, TableStyle
import qrcode
import json
import os
class CertificateGenerator:
def __init__(self, output_dir):
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
def generate_certificate_id(self):
return uuid.uuid4().hex
def generate_certificate_hash(self, data):
data_str = json.dumps(data, sort_keys=True)
return hashlib.sha256(data_str.encode()).hexdigest()
def create_certificate_data(self, student, institution, course, issue_date):
certificate_id = self.generate_certificate_id()
data = {
'certificate_id': certificate_id,
'student_name': student.name,
'student_id': student.id,
'institution_name': institution.name,
'institution_id': institution.id,
'course_name': course.name,
'course_id': course.id,
'issue_date': issue_date.isoformat(),
'timestamp': datetime.utcnow().isoformat()
}
hash_value = self.generate_certificate_hash(data)
data['hash'] = hash_value
return data, certificate_id, hash_value
def generate_pdf(self, data, qr_code_url=None):
certificate_id = data['certificate_id']
filename = f"{certificate_id}.pdf"
filepath = os.path.join(self.output_dir, filename)
c = canvas.Canvas(filepath, pagesize=letter)
width, height = letter
# 添加证书边框
c.setStrokeColor(colors.darkblue)
c.setLineWidth(3)
c.rect(30, 30, width - 60, height - 60, stroke=1, fill=0)
# 添加标题
c.setFont("Helvetica-Bold", 24)
c.drawCentredString(width/2, height - 100, "Certificate of Achievement")
# 添加学生姓名
c.setFont("Helvetica-Bold", 18)
c.drawCentredString(width/2, height - 180, f"This is to certify that")
c.setFont("Helvetica-Bold", 22)
c.drawCentredString(width/2, height - 220, data['student_name'])
# 添加课程详情
c.setFont("Helvetica", 16)
c.drawCentredString(width/2, height - 270, f"has successfully completed the course")
c.setFont("Helvetica-Bold", 18)
c.drawCentredString(width/2, height - 310, data['course_name'])
# 添加机构名称
c.setFont("Helvetica", 16)
c.drawCentredString(width/2, height - 360, f"offered by")
c.setFont("Helvetica-Bold", 18)
c.drawCentredString(width/2, height - 400, data['institution_name'])
# 添加日期
c.setFont("Helvetica", 14)
c.drawCentredString(width/2, height - 450, f"Issued on: {data['issue_date']}")
# 添加证书ID和哈希
c.setFont("Helvetica", 10)
c.drawString(50, 80, f"Certificate ID: {data['certificate_id']}")
c.drawString(50, 60, f"Verification Hash: {data['hash'][:20]}...")
# 添加二维码（如果提供）
if qr_code_url:
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=10,
border=4,
)
qr.add_data(qr_code_url)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
img_path = os.path.join(self.output_dir, f"{certificate_id}_qr.png")
img.save(img_path)
c.drawImage(img_path, width - 150, 50, width=100, height=100)
c.save()
return filepath

复制代码

证书验证

class CertificateValidator:
def __init__(self, ethereum_client):
self.ethereum_client = ethereum_client
def validate_certificate_file(self, certificate_file_path):
# 从PDF提取证书数据和哈希
certificate_data = self._extract_data_from_pdf(certificate_file_path)
if not certificate_data:
return {
'valid': False,
'error': 'Could not extract certificate data from file'
}
return self.validate_certificate_data(
certificate_data['certificate_id'],
certificate_data['hash']
)
def validate_certificate_data(self, certificate_id, hash_value):
# 在区块链上验证
blockchain_result = self.ethereum_client.verify_certificate(certificate_id, hash_value)
if not blockchain_result['is_valid']:
return {
'valid': False,
'error': 'Certificate not found or hash mismatch',
'details': blockchain_result
}
if blockchain_result['is_revoked']:
return {
'valid': False,
'error': 'Certificate has been revoked',
'details': blockchain_result
}
# 获取颁发者详情
issuer_address = blockchain_result['issuer']
issuer = self._get_issuer_by_address(issuer_address)
if not issuer:
return {
'valid': True,
'warning': 'Issuer not found in database',
'blockchain_details': blockchain_result
}
return {
'valid': True,
'issuer': {
'name': issuer.name,
'website': issuer.website
},
'issue_date': datetime.fromtimestamp(blockchain_result['timestamp']),
'blockchain_details': blockchain_result
}
def _extract_data_from_pdf(self, pdf_path):
# 从PDF提取证书数据的实现
# 这里会使用PyPDF2或类似库
pass
def _get_issuer_by_address(self, address):
# 通过以太坊地址获取机构的实现
# 这里会查询数据库
pass
### 5. Web接口实现
使用Flask构建Web接口，以下是主要路由的实现：
#### 认证路由
```python
from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from app.models import User
from app.forms import LoginForm, RegistrationForm
from app import db
auth_bp = Blueprint('auth', __name__)
@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = RegistrationForm()
if form.validate_on_submit():
user = User(
username=form.username.data,
email=form.email.data,
role=form.role.data
)
user.set_password(form.password.data)
db.session.add(user)
db.session.commit()
flash('恭喜，您已成功注册！')
return redirect(url_for('auth.login'))
return render_template('auth/register.html', title='注册', form=form)
@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user is None or not user.check_password(form.password.data):
flash('用户名或密码无效')
return redirect(url_for('auth.login'))
login_user(user, remember=form.remember_me.data)
next_page = request.args.get('next')
if not next_page or url_parse(next_page).netloc != '':
next_page = url_for('main.index')
return redirect(next_page)
return render_template('auth/login.html', title='登录', form=form)
@auth_bp.route('/logout')
def logout():
logout_user()
return redirect(url_for('main.index'))

复制代码

证书路由

from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify, send_file
from flask_login import login_required, current_user
from app.models import Certificate, User, Institution
from app.forms import CertificateForm, CertificateVerifyForm
from app.blockchain.ethereum import EthereumClient
from app.certificate.generator import CertificateGenerator
from app.certificate.validator import CertificateValidator
from app import db
import os
cert_bp = Blueprint('certificate', __name__)
@cert_bp.route('/issue', methods=['GET', 'POST'])
@login_required
def issue_certificate():
if current_user.role != 'institution':
flash('只有教育机构可以颁发证书')
return redirect(url_for('main.index'))
form = CertificateForm()
form.student.choices = [(u.id, u.username) for u in User.query.filter_by(role='student').all()]
if form.validate_on_submit():
student = User.query.get(form.student.data)
institution = Institution.query.filter_by(admin_id=current_user.id).first()
if not institution:
flash('您需要先设置机构信息')
return redirect(url_for('institution.setup'))
# 生成证书
generator = CertificateGenerator(output_dir='app/static/certificates')
certificate_data = {
'student_name': student.username,
'student_id': student.id,
'institution_name': institution.name,
'institution_id': institution.id,
'course_name': form.title.data,
'course_id': 'N/A',
'issue_date': form.issue_date.data.isoformat()
}
data, certificate_id, hash_value = generator.generate_certificate_data(certificate_data)
# 创建验证URL
verification_url = url_for('certificate.verify', certificate_id=certificate_id, _external=True)
# 生成PDF
pdf_path = generator.generate_pdf(data, verification_url)
relative_path = os.path.relpath(pdf_path, start='app')
# 存储到区块链
ethereum_client = EthereumClient(
provider_url=app.config['ETHEREUM_PROVIDER'],
contract_address=app.config['CONTRACT_ADDRESS'],
abi_path=app.config['CONTRACT_ABI_PATH'],
private_key=app.config['PRIVATE_KEY']
)
tx_hash = ethereum_client.issue_certificate(
certificate_id,
hash_value,
institution.ethereum_address
)
# 保存到数据库
certificate = Certificate(
certificate_id=certificate_id,
title=form.title.data,
description=form.description.data,
issue_date=form.issue_date.data,
expiry_date=form.expiry_date.data,
hash_value=hash_value,
blockchain_tx=tx_hash,
pdf_path=relative_path,
user_id=student.id,
institution_id=institution.id
)
db.session.add(certificate)
db.session.commit()
flash(f'证书颁发成功，ID: {certificate_id}')
return redirect(url_for('certificate.view', certificate_id=certificate_id))
return render_template('certificate/issue.html', title='颁发证书', form=form)
@cert_bp.route('/verify', methods=['GET', 'POST'])
def verify_certificate():
form = CertificateVerifyForm()
result = None
if form.validate_on_submit() or request.args.get('certificate_id'):
certificate_id = form.certificate_id.data or request.args.get('certificate_id')
# 从数据库获取证书
certificate = Certificate.query.filter_by(certificate_id=certificate_id).first()
if not certificate:
flash('数据库中未找到证书')
return render_template('certificate/verify.html', title='验证证书', form=form, result=None)
# 验证证书
ethereum_client = EthereumClient(
provider_url=app.config['ETHEREUM_PROVIDER'],
contract_address=app.config['CONTRACT_ADDRESS'],
abi_path=app.config['CONTRACT_ABI_PATH']
)
validator = CertificateValidator(ethereum_client)
result = validator.validate_certificate_data(certificate_id, certificate.hash_value)
# 添加数据库信息
if result['valid']:
result['certificate'] = {
'title': certificate.title,
'description': certificate.description,
'issue_date': certificate.issue_date,
'student': User.query.get(certificate.user_id).username,
'institution': Institution.query.get(certificate.institution_id).name
}
return render_template('certificate/verify.html', title='验证证书', form=form, result=result)
@cert_bp.route('/view/<certificate_id>')
def view_certificate(certificate_id):
certificate = Certificate.query.filter_by(certificate_id=certificate_id).first_or_404()
# 检查权限
if certificate.user_id != current_user.id and certificate.institution.admin_id != current_user.id and current_user.role != 'admin':
flash('您无权查看此证书')
return redirect(url_for('main.index'))
return render_template('certificate/view.html', title='查看证书', certificate=certificate)
@cert_bp.route('/download/<certificate_id>')
def download_certificate(certificate_id):
certificate = Certificate.query.filter_by(certificate_id=certificate_id).first_or_404()
# 检查权限
if certificate.user_id != current_user.id and certificate.institution.admin_id != current_user.id and current_user.role != 'admin':
flash('您无权下载此证书')
return redirect(url_for('main.index'))
return send_file(os.path.join('app', certificate.pdf_path), as_attachment=True)
### 6. 系统部署
#### 配置文件
```python
# config.py
import os
from dotenv import load_dotenv
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess-string'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(basedir, 'app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 以太坊配置
ETHEREUM_PROVIDER = os.environ.get('ETHEREUM_PROVIDER') or 'http://localhost:8545'
CONTRACT_ADDRESS = os.environ.get('CONTRACT_ADDRESS')
CONTRACT_ABI_PATH = os.path.join(basedir, 'contract_abi.json')
PRIVATE_KEY = os.environ.get('PRIVATE_KEY')

复制代码

依靠文件

# requirements.txt
flask==2.0.1
flask-sqlalchemy==2.5.1
flask-migrate==3.1.0
flask-login==0.5.0
flask-wtf==0.15.1
web3==5.24.0
python-dotenv==0.19.0
pycryptodome==3.10.1
reportlab==3.6.1
qrcode==7.3
PyPDF2==1.26.0
gunicorn==20.1.0

复制代码

启动脚本

# run.py
from app import create_app, db
from app.models import User, Institution, Certificate
app = create_app()
@app.shell_context_processor
def make_shell_context():
return {
'db': db,
'User': User,
'Institution': Institution,
'Certificate': Certificate
}
if __name__ == '__main__':
app.run(debug=True)

复制代码

体系特点与上风

安全性

区块链不可篡改性：一旦证书信息被记录在区块链上，就无法被篡改
密码学验证：利用哈希函数确保证书内容的完备性
权限控制：严格的用户角色和权限管理

透明性

公开验证：任何人都可以验证证书的真实性
完备审计追踪：全部证书颁发和验证操纵都有记录
机构信誉可查：可以查察教育机构颁发的全部证书

可用性

用户友爱界面：简单直观的操纵流程
多平台支持：Web界面适配各种设备
离线验证选项：通过二维码提供离线验证功能

未来预测

功能扩展

多语言支持：添加多语言界面，支持国际化
批量证书颁发：支持一次性为多名门生颁发证书
证书模板体系：答应机构自定义证书模板

技能升级

跨链集成：支持多种区块链平台
零知识证明：增强隐私保护功能
AI辅助验证：利用AI技能增强证书验证过程

生态体系扩展

API接口：提供API接供词第三方体系集成
移动应用：开辟配套移动应用
人才招聘平台集成：与招聘平台集成，简化学历验证流程

结论

基于区块链的门生证书认证体系为解决学历造假题目提供了一种创新的解决方案。通过结合区块链技能的不可篡改性和传统数据库的高效查询本领，该体系既保证了证书的真实性和可信度，又提供了良好的用户体验。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

Python项目--基于区块链的门生证书认证体系

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块

		自动登录	找回密码
密码			立即注册