1. 项目概述与核心需求
这个Python项目模拟了一个常见的用户登录验证场景:当用户连续三次输入错误密码时,系统会自动锁定该账号。这种机制在各类网站和应用中非常普遍,是基础的安全防护措施之一。
核心功能需求可以分解为:
- 用户输入用户名和密码进行验证
- 允许最多三次错误尝试
- 超过限制后将用户名加入黑名单
- 后续登录时检查黑名单并阻止登录
2. 基础实现方案解析
2.1 基础代码结构
我们先来看一个最基础的实现版本:
# 基础配置 MAX_ATTEMPTS = 3 BLACKLIST_FILE = 'blacklist.txt' CORRECT_USERNAME = 'admin' CORRECT_PASSWORD = '123456' # 检查黑名单 def check_blacklist(username): try: with open(BLACKLIST_FILE, 'r') as f: return username in f.read().splitlines() except FileNotFoundError: return False # 主登录逻辑 def login(): username = input("请输入用户名: ") if check_blacklist(username): print("账号已被锁定,请联系管理员") return attempts = 0 while attempts < MAX_ATTEMPTS: password = input("请输入密码: ") if username == CORRECT_USERNAME and password == CORRECT_PASSWORD: print("登录成功!") return else: attempts += 1 remaining = MAX_ATTEMPTS - attempts print(f"密码错误,剩余尝试次数: {remaining}") # 超过尝试次数 with open(BLACKLIST_FILE, 'a') as f: f.write(f"{username}\n") print("错误次数过多,账号已被锁定") if __name__ == '__main__': login()2.2 关键点解析
黑名单机制:
- 使用文本文件存储被锁定的用户名
- 每次登录前先检查黑名单
- 简单但存在并发问题(后面会讨论改进方案)
尝试次数控制:
- 使用while循环控制最大尝试次数
- 实时显示剩余尝试次数提升用户体验
密码验证逻辑:
- 直接比较字符串(实际应用中应该使用哈希比较)
- 用户名和密码硬编码在代码中(实际应该从数据库读取)
3. 进阶实现与优化
3.1 使用哈希密码存储
实际应用中,密码绝不能明文存储。我们引入hashlib进行密码哈希:
import hashlib def hash_password(password): return hashlib.sha256(password.encode()).hexdigest() # 配置改为存储哈希值 CORRECT_PASSWORD_HASH = hash_password("123456") # 验证时比较哈希值 if username == CORRECT_USERNAME and hash_password(password) == CORRECT_PASSWORD_HASH: print("登录成功!")3.2 添加账户锁定时间
当前实现永久锁定账户不合理,应该设置锁定时间:
import time import json LOCK_DURATION = 3600 # 锁定1小时 def add_to_blacklist(username): lock_data = { 'username': username, 'lock_time': time.time() } with open(BLACKLIST_FILE, 'a') as f: f.write(json.dumps(lock_data) + '\n') def check_blacklist(username): try: with open(BLACKLIST_FILE, 'r') as f: for line in f: data = json.loads(line) if data['username'] == username: if time.time() - data['lock_time'] < LOCK_DURATION: return True else: return False return False except FileNotFoundError: return False3.3 使用数据库存储
文本文件不适合生产环境,改用SQLite:
import sqlite3 import hashlib import time def init_db(): conn = sqlite3.connect('auth.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS users (username TEXT PRIMARY KEY, password_hash TEXT)''') c.execute('''CREATE TABLE IF NOT EXISTS blacklist (username TEXT PRIMARY KEY, lock_time REAL)''') # 添加测试用户 c.execute("INSERT OR IGNORE INTO users VALUES (?, ?)", ('admin', hash_password('123456'))) conn.commit() conn.close() def check_credentials(username, password): conn = sqlite3.connect('auth.db') c = conn.cursor() c.execute("SELECT password_hash FROM users WHERE username=?", (username,)) result = c.fetchone() conn.close() return result and result[0] == hash_password(password)4. 安全增强措施
4.1 防止暴力破解
简单的尝试次数限制还不够,还需要:
- 增加延迟:每次失败后增加等待时间
import time def login(): # ... for attempt in range(MAX_ATTEMPTS): password = input("请输入密码: ") if check_credentials(username, password): print("登录成功!") return else: wait_time = (attempt + 1) * 2 # 线性增加等待时间 print(f"密码错误,请等待{wait_time}秒后重试") time.sleep(wait_time)- IP限制:记录IP的失败尝试
4.2 密码安全策略
- 密码复杂度要求
- 定期强制修改密码
- 禁止使用近期用过的密码
5. 常见问题与解决方案
5.1 并发访问问题
原始实现存在多个问题:
- 多个进程同时写入黑名单文件可能损坏数据
- 尝试次数计数不准确
解决方案:
- 使用文件锁(fcntl或portalocker)
- 使用数据库事务
- 使用专门的缓存系统如Redis
import portalocker def add_to_blacklist(username): with portalocker.Lock(BLACKLIST_FILE, 'a') as f: f.write(f"{username}\n")5.2 用户体验优化
- 显示剩余尝试次数
- 友好的错误提示
- 密码输入隐藏(使用getpass模块)
from getpass import getpass password = getpass("请输入密码: ")5.3 日志记录
添加详细的日志记录:
import logging logging.basicConfig(filename='auth.log', level=logging.INFO) def login(): username = input("用户名: ") logging.info(f"登录尝试: {username}") if check_blacklist(username): logging.warning(f"尝试登录已锁定账号: {username}") print("账号已被锁定") return # ...其余登录逻辑...6. 完整实现示例
结合所有优化后的完整代码:
import sqlite3 import hashlib import time import logging from getpass import getpass # 配置 MAX_ATTEMPTS = 3 LOCK_DURATION = 3600 # 1小时 DATABASE = 'auth.db' # 初始化日志 logging.basicConfig( filename='auth.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) def hash_password(password): return hashlib.sha256(password.encode()).hexdigest() def init_db(): conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS users (username TEXT PRIMARY KEY, password_hash TEXT, last_login REAL)''') c.execute('''CREATE TABLE IF NOT EXISTS login_attempts (id INTEGER PRIMARY KEY, username TEXT, attempt_time REAL, ip_address TEXT)''') c.execute('''CREATE TABLE IF NOT EXISTS blacklist (username TEXT PRIMARY KEY, lock_time REAL)''') # 添加测试用户 test_hash = hash_password('123456') c.execute("INSERT OR IGNORE INTO users VALUES (?, ?, ?)", ('admin', test_hash, 0)) conn.commit() conn.close() def check_credentials(username, password): conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute("SELECT password_hash FROM users WHERE username=?", (username,)) result = c.fetchone() conn.close() return result and result[0] == hash_password(password) def is_locked(username): conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute("SELECT lock_time FROM blacklist WHERE username=?", (username,)) result = c.fetchone() conn.close() if result: lock_time = result[0] if time.time() - lock_time < LOCK_DURATION: return True else: # 锁定已过期,移除记录 conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute("DELETE FROM blacklist WHERE username=?", (username,)) conn.commit() conn.close() return False return False def record_attempt(username, ip='127.0.0.1'): conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute("INSERT INTO login_attempts VALUES (NULL, ?, ?, ?)", (username, time.time(), ip)) conn.commit() conn.close() def get_recent_attempts(username, minutes=10): conn = sqlite3.connect(DATABASE) c = conn.cursor() cutoff = time.time() - minutes * 60 c.execute("""SELECT COUNT(*) FROM login_attempts WHERE username=? AND attempt_time > ?""", (username, cutoff)) result = c.fetchone() conn.close() return result[0] if result else 0 def lock_account(username): conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute("INSERT OR REPLACE INTO blacklist VALUES (?, ?)", (username, time.time())) conn.commit() conn.close() logging.warning(f"账号锁定: {username}") def login(): username = input("用户名: ").strip() logging.info(f"登录尝试: {username}") if is_locked(username): print("该账号已被锁定,请稍后再试或联系管理员") return recent_attempts = get_recent_attempts(username) if recent_attempts >= MAX_ATTEMPTS: lock_account(username) print("尝试次数过多,账号已被临时锁定") return for attempt in range(MAX_ATTEMPTS - recent_attempts): password = getpass("密码: ") if check_credentials(username, password): print("登录成功!") # 更新最后登录时间 conn = sqlite3.connect(DATABASE) c = conn.cursor() c.execute("UPDATE users SET last_login=? WHERE username=?", (time.time(), username)) conn.commit() conn.close() return else: record_attempt(username) remaining = MAX_ATTEMPTS - recent_attempts - attempt - 1 if remaining > 0: print(f"密码错误,剩余尝试次数: {remaining}") # 增加延迟 time.sleep((attempt + 1) * 2) else: lock_account(username) print("错误次数过多,账号已被锁定") if __name__ == '__main__': init_db() login()7. 项目扩展方向
7.1 添加Web界面
使用Flask或Django将代码转换为Web应用:
from flask import Flask, request, jsonify app = Flask(__name__) @app.route('/login', methods=['POST']) def web_login(): data = request.get_json() username = data.get('username') password = data.get('password') if is_locked(username): return jsonify({'status': 'error', 'message': 'Account locked'}), 403 if check_credentials(username, password): return jsonify({'status': 'success'}) else: record_attempt(username, request.remote_addr) return jsonify({'status': 'error', 'message': 'Invalid credentials'}), 4017.2 多因素认证
增加短信或邮箱验证码:
- 集成短信API
- 生成随机验证码
- 验证时检查验证码
7.3 风险评估系统
基于以下因素评估登录风险:
- 登录地理位置
- 登录设备
- 登录时间
- 行为模式
对高风险登录要求额外验证
8. 测试与验证
8.1 单元测试
使用unittest编写测试用例:
import unittest import os class TestAuthSystem(unittest.TestCase): @classmethod def setUpClass(cls): # 使用内存数据库测试 global DATABASE DATABASE = ':memory:' init_db() def test_successful_login(self): # 模拟输入 # 需要重写input和getpass函数 global input, getpass input = lambda _: 'admin' getpass = lambda _: '123456' # 捕获打印输出 from io import StringIO import sys saved_stdout = sys.stdout try: out = StringIO() sys.stdout = out login() output = out.getvalue().strip() self.assertIn("登录成功", output) finally: sys.stdout = saved_stdout def test_failed_login(self): # 类似地测试失败情况 pass if __name__ == '__main__': unittest.main()8.2 集成测试
测试整个流程:
- 正常登录
- 错误密码登录
- 锁定后尝试登录
- 锁定过期后登录
9. 部署注意事项
数据库安全:
- 使用适当的数据库权限
- 定期备份
- 加密敏感数据
日志管理:
- 设置日志轮转
- 监控异常登录尝试
- 保护日志文件不被未授权访问
性能考虑:
- 数据库连接池
- 缓存常用查询
- 负载测试
10. 实际应用中的挑战
密码重置流程:
- 安全问题验证
- 邮件/SMS验证链接
- 临时密码机制
用户自服务:
- 解锁账号流程
- 密码强度检查
- 安全提示设置
审计合规:
- 记录所有关键操作
- 不可否认性
- 定期审计日志
这个项目虽然从表面看只是一个简单的登录验证,但深入实现后会涉及到许多安全、用户体验和系统设计方面的考虑。在实际开发中,建议使用成熟的认证框架如Django的认证系统或Auth0等专业服务,它们已经处理了大多数边缘情况和安全问题。