创新实训(二)——FastAPI后端登录注册功能实现及前后端连接
登录注册功能是保护用户安全性的第一道防线,本次任务中,我的目标是构建一个完整的用户认证系统,实现后端的登录注册功能,结合 SQLModel 进行数据库操作,并完成前后端的数据交互。此功能是后续业务模块(如数据管理、权限控制)的基础,需确保系统安全性和可扩展性。
目录
登录注册逻辑流程
安全性保障技术
1. bcrypt 算法密码加密
2. JWT(JSON Web Token)身份验证
3. 图形验证码
4.邮箱验证码
前后端连接
登录注册逻辑流程
登录注册功能中,为保证各方安全性,在登录环节中,设置数字图像验证码来防止自动化攻击,保护系统的安全性;在注册环节中,设置邮箱验证码,必须在对应邮箱收到验证码进行验证,才能完成注册,保证了用户账户的安全。本系统的登陆注册流程图如下图所示:
安全性保障技术
1.bcrypt算法密码加密
bcrypt算法属于慢哈希算法,能够有效抵御暴力破解攻击。同时,它还具备自动加盐的功能,进一步增强了密码的安全性。在用户注册时,系统会对输入的密码进行加密处理,然后再将加密后的密码存储到数据库中,避免了明文密码存储带来的安全风险。
from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def get_password_hash(password: str) -> str: return pwd_context.hash(password)2.JWT(JSON Web Token)身份验证
JWT 是一种基于 JSON 的开放标准,用于创建访问令牌,允许在各方之间安全地传输信息。JWT的声明一般被用来在身份提供者和服务提供者间传递被认证的用户身份信息,以便于从资源服务器获取资源,也可以增加一些额外的其它业务逻辑所必须的声明信息,该token也可直接被用于认证,也可被加密。此外,JWT 令牌包含了过期时间,过期后将无法使用,有效防止了令牌被滥用。
下图为登录时后端返回给前端的token
from datetime import datetime, timedelta, timezone from jose import jwt #生成token def create_access_token(subject: str | Any, expires_delta: timedelta) -> str: expire = datetime.now(timezone.utc) + expires_delta to_encode = {"exp": expire, "sub": str(subject)} encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt3. 图形验证码
为保证系统的安全性,预防恶意攻击,在登录时要求用户输入图形验证码进行验证。实现效果如下图所示:
创建CaptchaService类用于生成图形验证码,它会生成包含随机数学运算题目的图片,并将图片数据转换为 Base64 字符串和问题的答案,代码如下:
class CaptchaService: @classmethod async def create_captcha_image_service(cls): # 创建空白图像 image = Image.new('RGB', (160, 60), color='#EAEAEA') # 创建绘图对象 draw = ImageDraw.Draw(image) # 设置字体 font = ImageFont.truetype(os.path.join(os.path.abspath(os.getcwd()), 'assets', 'font', 'Arial.ttf'), size=30) # 生成两个0-9之间的随机整数 num1 = random.randint(0, 9) num2 = random.randint(0, 9) # 从运算符列表中随机选择一个 operational_character_list = ['+', '-', '*'] operational_character = random.choice(operational_character_list) # 根据选择的运算符进行计算 if operational_character == '+': result = num1 + num2 elif operational_character == '-': result = num1 - num2 else: result = num1 * num2 # 绘制文本 text = f'{num1} {operational_character} {num2} = ?' draw.text((25, 15), text, fill='blue', font=font) # 将图像数据保存到内存中 buffer = io.BytesIO() image.save(buffer, format='PNG') # 将图像数据转换为base64字符串 base64_string = base64.b64encode(buffer.getvalue()).decode() return [base64_string, result]当前端发起获取图像验证码的请求时,后端接口调用此方法生成图像验证码,并把验证码和答案一起传到前端。
4.邮箱验证码
EmailCodeService类用于生成和管理邮箱验证码。它会生成 6 位数字验证码,并将验证码存储在内存中,用于后续根据用户邮箱进行验证,同时设置有效期为 5 分钟。
实现效果如下图所示:
在邮箱验证码功能的实现中,首先为目的邮箱生成验证码,并存储在内存中;在邮箱验证类中,设置过期时间为5分钟,并在每次生成验证码的时候清除已过期的验证码,防止占据太多内存。在发送邮箱前,根据提前准备好的模板,完善邮箱内容,传入send_email方法中进行发送。
@router.post("/getEmailCode") async def send_email_code(request: EmailCodeRequest): # 生成并存储验证码 code = EmailCodeService.generate_code() EmailCodeService.store_code(request.email, code) # 清理过期验证码 EmailCodeService.cleanup_expired() # 发送邮件 email_data = generate_email_code_template(email_to=request.email, code=code) try: send_email( email_to=request.email, subject=email_data.subject, html_content=email_data.html_content ) except Exception as e: logger.error(f"邮件发送失败: {str(e)}") raise HTTPException(503, detail=str(e)) return {"code": 0}在发送邮件的过程中, 需要获取邮箱的相关配置信息,这些信息在Setting类中定义。为保护开发者信息的安全性,我选择将具体配置信息存在.env文件中,并在Setting类中获取该文件里的内容。
def send_email( *, email_to: str, subject: str = "", html_content: str = "", ) -> None: message = emails.Message( subject=subject, html=html_content, mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL), ) try: smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT} if settings.SMTP_TLS: smtp_options["tls"] = True elif settings.SMTP_SSL: smtp_options["ssl"] = True if settings.SMTP_USER: smtp_options["user"] = settings.SMTP_USER if settings.SMTP_PASSWORD: smtp_options["password"] = settings.SMTP_PASSWORD response = message.send(to=email_to, smtp=smtp_options) logger.info(f"send email result: {response}") except Exception as e: print({"detail": f"邮件发送失败: {str(e)}", "traceback": traceback.format_exc()})前后端连接
本次连接是该项目的第一次前后端连接,前端使用vue3,后端使用FastAPI。前后端连接其实就是前端向后端发起请求,后端对请求进行响应,并将响应结果发回前端的过程,在之前的前后端分离项目中也多次涉及到。
对于FastAPI的前后端连接,在前后端分离的架构下,由于前端和后端可能运行在不同的域名或端口下,这就需要处理跨域问题。首先要在后端导入中间件CORSMiddleware(跨域资源共享中间件),在allow_origins中指定了允许访问后端 API 的前端来源。
app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:4000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )随后在前端,根据后端提供的api发起请求,以登录方法为例:
定义userLoginReq的异步函数,它接收两个字符串类型的参数username和password,用于传递用户输入的登录信息,函数返回一个Promise对象,表明这是一个异步操作。
在 try 代码块中,通过 axios 库发起一个POST请求,请求的目标URL是 '/auth/users/login',这是后端定义的用于处理用户登录请求的接口地址。
对于需要传输的数据,运用qs.stringify 方法,它将包含username和password的对象转换为 application/x-www-form-urlencoded 格式的字符串,即 username=xxx&password=yyy 这种形式。这是因为后端接口预期接收的是这种格式的数据,同时,代码中明确设置了请求头 'Content-Type': 'application/x-www-form-urlencoded',以告知服务器请求体的数据格式。
最后,根据后端响应的状态码和方法体中捕获到的错误,及时将错误向用户反馈。
export async function userLoginReq(username: string, password: string): Promise<any> { try { const res = await axios.post('/auth/users/login', qs.stringify({ username: username, password: password }), { headers: { 'Content-Type': 'application/x-www-form-urlencoded' } } ) //等待从服务器返还 if ( res.status == 200 && res.data.access_token != null && res.data.access_token != '' && res.data.access_token != undefined ) { return res.data } else { throw new Error('未知错误'); } } catch (error: any) { const errorMessage = error.response?.data?.detail || '登录失败,请检查网络连接'; throw new Error(`登录失败: ${errorMessage}`); } }