Python实现用户登录验证与账号锁定机制
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 使用数据库存储文本文件不适合生产环境改用SQLiteimport 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使用数据库事务使用专门的缓存系统如Redisimport 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(filenameauth.log, levellogging.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( filenameauth.log, levellogging.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, ip127.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, minutes10): 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等专业服务它们已经处理了大多数边缘情况和安全问题。