网站导航
> 技术文档 > Python项目--基于区块链的学生证书认证系统_基于区块链的证书管理系统

Python项目--基于区块链的学生证书认证系统_基于区块链的证书管理系统

网友: 技术文档 2025-07-30 12:20:40

引言

在当今数字化时代，学历造假和证书伪造已成为教育领域的一大挑战。传统的证书验证方法往往耗时且容易被篡改，这促使我们寻求更加安全、透明和高效的解决方案。区块链技术以其不可篡改、去中心化和透明的特性，为学生证书认证提供了理想的技术基础。

本文将详细介绍如何使用Python和区块链技术构建一个完整的学生证书认证系统，该系统允许教育机构安全地颁发数字证书，同时使学生和雇主能够轻松验证这些证书的真实性。

系统概述

我们的基于区块链的学生证书认证系统主要包含以下几个核心组件：

  1. 区块链网络：用于存储证书哈希和元数据
  2. 证书生成器：为学生创建数字证书
  3. 证书验证器：验证证书的真实性
  4. Web界面：用户友好的前端界面
  5. 数据库：存储用户信息和证书详情

系统工作流程如下：

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

技术栈

本项目使用以下技术栈：

  • 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: MITpragma 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 Web3import jsonclass 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 hashlibimport uuidfrom datetime import datetimefrom reportlab.lib.pagesizes import letterfrom reportlab.pdfgen import canvasfrom reportlab.lib import colorsfrom reportlab.lib.styles import getSampleStyleSheetfrom reportlab.platypus import Paragraph, Table, TableStyleimport qrcodeimport jsonimport osclass 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接口，以下是主要路由的实现：#### 认证路由```pythonfrom flask import Blueprint, render_template, redirect, url_for, flash, requestfrom flask_login import login_user, logout_user, login_required, current_userfrom app.models import Userfrom app.forms import LoginForm, RegistrationFormfrom app import dbauth_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_filefrom flask_login import login_required, current_userfrom app.models import Certificate, User, Institutionfrom app.forms import CertificateForm, CertificateVerifyFormfrom app.blockchain.ethereum import EthereumClientfrom app.certificate.generator import CertificateGeneratorfrom app.certificate.validator import CertificateValidatorfrom app import dbimport oscert_bp = Blueprint(\'certificate\', __name__)@cert_bp.route(\'/issue\', methods=[\'GET\', \'POST\'])@login_requireddef 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/\')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/\')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.pyimport osfrom dotenv import load_dotenvbasedir = 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.txtflask==2.0.1flask-sqlalchemy==2.5.1flask-migrate==3.1.0flask-login==0.5.0flask-wtf==0.15.1web3==5.24.0python-dotenv==0.19.0pycryptodome==3.10.1reportlab==3.6.1qrcode==7.3PyPDF2==1.26.0gunicorn==20.1.0
启动脚本
# run.pyfrom app import create_app, dbfrom app.models import User, Institution, Certificateapp = create_app()@app.shell_context_processordef make_shell_context(): return { \'db\': db, \'User\': User, \'Institution\': Institution, \'Certificate\': Certificate }if __name__ == \'__main__\': app.run(debug=True)

系统特点与优势

安全性

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

透明性

  • 公开验证：任何人都可以验证证书的真实性
  • 完整审计追踪：所有证书颁发和验证操作都有记录
  • 机构信誉可查：可以查看教育机构颁发的所有证书

可用性

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

未来展望

功能扩展

  1. 多语言支持：添加多语言界面，支持国际化
  2. 批量证书颁发：支持一次性为多名学生颁发证书
  3. 证书模板系统：允许机构自定义证书模板

技术升级

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

生态系统扩展

  1. API接口：提供API接口供第三方系统集成
  2. 移动应用：开发配套移动应用
  3. 人才招聘平台集成：与招聘平台集成，简化学历验证流程

结论

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

网络标签：

相关问题