6.7 CRM 微服务实战（三）：认证授权与中间件

引言：保护我们的服务

在上一章，我们为 CRM 微服务实现了核心的业务逻辑。它现在功能完备，可以对联系人进行增删改查。但是，它还存在一个巨大的问题：它完全是“裸奔”的。任何能够连接到我们服务端口的人，都可以随意地创建、读取、甚至删除所有数据。

在一个真实的系统中，这是绝对不能接受的。我们需要一个坚固的“大门”来保护我们的服务。这个大门至少要做两件事：

  1. 认证 (Authentication)：验证“你是谁”。这通常通过检查请求中携带的凭证（如 API Key, JWT, 用户名/密码）来完成。
  2. 授权 (Authorization)：验证“你能做什么”。在确认了你的身份后，还需要检查你是否有权限执行当前的操作（例如，普通用户只能查看自己的联系人，而管理员可以查看所有人的）。

手动在每个 gRPC 方法的开头都添加认证和授权的检查代码是重复且易错的。这正是中间件大显身手的舞台。正如我们在 axum 和 tower 章节学到的，我们可以使用 tower::Layer 来实现这些横切关注点，将它们与核心业务逻辑完全解耦。

本章，我们将为我们的 CRM 服务构建一个完整的认证和授权层。我们将学习：

  • 如何使用 tonic 的拦截器实现简单的 API Key 认证。
  • 如何编写一个自定义的 tower 中间件，从数据库中加载用户信息并进行授权。
  • 如何将用户身份信息安全地传递给我们的业务逻辑 handler。

1. 简单的 API Key 认证：使用拦截器

对于内部微服务间的通信，一种常见的认证方式是使用预共享的 API Key。每个合法的客户端在发起请求时，都需要在元数据中携带这个 Key。

tonic 的拦截器 (Interceptor) 非常适合实现这种简单的、只读请求元数据的认证逻辑。

实现认证拦截器

让我们创建一个 auth.rs 模块来存放我们的认证逻辑。
src/auth.rs:

use tonic::{Request, Status};

const SECRET_API_KEY: &str = "my-secret-crm-key";

pub fn auth_interceptor(mut req: Request<()>) -> Result<Request<()>, Status> {
    // 从元数据中查找 "x-api-key"
    let api_key = match req.metadata().get("x-api-key") {
        Some(key) => key.to_str(),
        None => return Err(Status::unauthenticated("Missing API key")),
    };

    if let Ok(key) = api_key {
        if key == SECRET_API_KEY {
            // Key 有效，继续请求
            Ok(req)
        } else {
            Err(Status::unauthenticated("Invalid API key"))
        }
    } else {
        Err(Status::unauthenticated("Invalid API key format"))
    }
}

这个拦截器函数非常简单：它检查 x-api-key 元数据是否存在且是否等于我们预设的密钥。如果不是，它就直接返回一个 Unauthenticated 状态，请求处理链会在此中断，不会到达我们的业务逻辑。

在服务器上应用拦截器

在 main.rs 中，我们可以使用 .with_interceptor() 将它应用到我们的服务上。

src/main.rs:

mod auth;
// ...

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ...
    let crm_service = pb::crm_server::CrmServer::new(service::CrmService::new(state))
        .with_interceptor(auth::auth_interceptor); // <--- 应用拦截器

    Server::builder()
        .add_service(crm_service)
        .serve(addr)
        .await?;
    // ...
}

现在，所有对 CrmService 的调用都必须先通过 auth_interceptor 的检查。

你可以修改上一章的 client.rs，在创建 Request 后，使用 request.metadata_mut().insert("x-api-key", "...") 来添加 API Key 并进行测试。

2. 数据库驱动的授权：自定义 Layer

简单的 API Key 认证能区分“合法的内部服务”和“非法的调用者”，但它无法区分不同的用户和不同的权限。例如，我们希望：

  • 用户 A 只能 list_contacts 属于他自己公司（customer_id）的联系人。
  • 一个“管理员”用户则可以访问所有联系人。

要实现这一点，我们需要：

  1. 在请求中包含用户身份信息（例如，一个 JWT，或者在本例中我们简化为一个用户 ID）。
  2. 在中间件中，根据用户 ID 从数据库中查询该用户的详细信息（包括他的角色和所属的 customer_id）。
  3. 将这些用户信息“注入”到请求中，以便后续的 handler 使用。
  4. （更进一步）在中间件中直接进行一些粗粒度的授权检查。

这种逻辑比简单的元数据检查要复杂，并且需要访问数据库（一个异步操作），因此使用完整的 tower::Layer 是更合适的选择。

定义用户身份和中间件 Layer

首先，我们定义一个结构体来表示通过认证的用户身份，这个结构体将在请求的生命周期中被传递。

src/auth.rs:

use crate::AppState;
use sqlx::PgPool;

#[derive(Debug, Clone)]
pub struct UserInfo {
    pub id: i64,
    pub customer_id: i64,
    pub roles: Vec<String>,
}

// 我们的授权 Layer
#[derive(Clone)]
pub struct AuthLayer {
    pool: PgPool,
}

impl AuthLayer {
    pub fn new(pool: PgPool) -> Self {
        Self { pool }
    }
}

接下来，实现 Layer 和 Service trait。

// src/auth.rs
use tower::{Layer, Service};
use std::future::Future;
use std::pin::Pin;
use hyper::Request as HyperRequest;
use hyper::body::Body;

impl<S> Layer<S> for AuthLayer {
    type Service = AuthService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        AuthService {
            inner,
            pool: self.pool.clone(),
        }
    }
}

#[derive(Clone)]
pub struct AuthService<S> {
    inner: S,
    pool: PgPool,
}

// 为 AuthService 实现 tower::Service
impl<S> Service<HyperRequest<Body>> for AuthService<S>
where
    S: Service<HyperRequest<Body>> + Clone + Send + 'static,
    S::Future: Send + 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
    
    fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: HyperRequest<Body>) -> Self::Future {
        let pool = self.pool.clone();
        let mut inner = self.inner.clone();

        Box::pin(async move {
            // 1. 从请求元数据中提取用户 ID
            // 在真实世界中，这里会解析 JWT token
            let user_id_str = match req.metadata().get("x-user-id") {
                Some(v) => v.to_str().unwrap_or(""),
                None => {
                    // 如果没有提供用户 ID，我们也可以允许匿名访问或直接拒绝
                    // 这里我们选择拒绝
                    let status = Status::unauthenticated("Missing user ID");
                    let response = Self::status_to_response(status);
                    return Ok(response);
                }
            };
            let user_id: i64 = match user_id_str.parse() {
                Ok(id) => id,
                Err(_) => {
                    let status = Status::invalid_argument("Invalid user ID format");
                    return Ok(Self::status_to_response(status));
                }
            };

            // 2. 异步地从数据库查询用户信息
            // 假设我们有一个 `auth_users` 表
            let user_info: Result<UserInfo, sqlx::Error> = sqlx::query_as!(
                UserInfo,
                "SELECT id, customer_id, roles FROM auth_users WHERE id = $1",
                user_id
            )
            .fetch_one(&pool)
            .await;

            match user_info {
                Ok(info) => {
                    // 3. 将用户信息注入到请求的 extensions 中
                    req.extensions_mut().insert(info);
                    // 4. 调用内部服务
                    inner.call(req).await
                }
                Err(sqlx::Error::RowNotFound) => {
                    let status = Status::unauthenticated("User not found");
                    Ok(Self::status_to_response(status))
                }
                Err(_) => {
                    let status = Status::internal("Internal server error during authentication");
                    Ok(Self::status_to_response(status))
                }
            }
        })
    }
}

// 辅助函数，将 gRPC Status 转换成一个 HTTP/2 响应
// tonic 在底层会做类似的事情，但我们的中间件在 tonic 服务之前，所以需要自己处理
impl<S> AuthService<S> {
     fn status_to_response(status: Status) -> hyper::Response<Body> {
        let mut res = hyper::Response::new(Body::empty());
        *res.status_mut() = hyper::StatusCode::OK;
        res.headers_mut().insert("grpc-status", status.code().to_string().parse().unwrap());
        if let Some(msg) = status.message().parse().ok() {
            res.headers_mut().insert("grpc-message", msg);
        }
        res
     }
}

我们需要创建一个 auth_users 表来测试这个中间件。

3. 在 Handler 中提取和使用用户身份

现在，中间件已经将 UserInfo 注入到了请求的 extensions 中。我们的 gRPC 方法需要一种方式来访问它。我们可以通过 Request::extensions() 来手动获取，但更优雅的方式是为 UserInfo 实现 FromRequestParts，让它成为一个提取器。

不过，tonic 的 RPC 方法签名是固定的 (&self, request: Request<T>)，它不像 axum 的 handler 那样可以直接在参数中添加提取器。

所以，我们需要在每个方法的开头从 request 中提取 extensions。

// src/service.rs
use crate::auth::UserInfo; // 引入我们的 UserInfo

// ...

#[tonic::async_trait]
impl Crm for CrmService {
    async fn get_contact(
        &self,
        request: Request<GetContactRequest>,
    ) -> Result<Response<PbContact>, Status> {
        // 1. 从请求中提取 UserInfo
        let user_info = request.extensions().get::<UserInfo>()
            .ok_or_else(|| Status::unauthenticated("No user info in request"))?;
        
        // 如果我们想记录是谁在操作
        tracing::info!("User {} is requesting contact {}", user_info.id, request.get_ref().id);

        let req = request.into_inner();

        let contact = sqlx::query_as!(
            Contact,
            "SELECT * FROM contacts WHERE id = $1",
            req.id
        )
        .fetch_one(&self.state.db_pool)
        .await
        .map_err(|e| match e {
            sqlx::Error::RowNotFound => Status::not_found(format!("Contact with id {} not found", req.id)),
            _ => Status::internal("Failed to get contact"),
        })?;

        // 2. 在这里进行授权检查
        // 普通用户只能查看自己公司的联系人
        if !user_info.roles.contains(&"admin".to_string()) && user_info.customer_id != contact.customer_id {
            return Err(Status::permission_denied("You do not have permission to view this contact"));
        }

        Ok(Response::new(contact.into()))
    }
    
    // ... 其他方法也需要类似的修改 ...
}

4. 在服务器上应用 AuthLayer

main.rs:

// ...
mod auth;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // ... (state 初始化) ...
    
    let auth_layer = auth::AuthLayer::new(state.db_pool.clone());
    
    let crm_service = pb::crm_server::CrmServer::new(service::CrmService::new(state));
    
    // 使用 ServiceBuilder 组合 Layer 和 Service
    let service = ServiceBuilder::new()
        .layer(TraceLayer::new_for_grpc())
        .layer(auth_layer) // 应用我们的 AuthLayer
        .service(crm_service);
        
    Server::builder()
        .add_service(service) // 添加被包装过的服务
        .serve(addr)
        .await?;

    Ok(())
}

现在，我们的服务拥有了一个强大的、由数据库驱动的认证授权层。每个请求都会被检查，并且 handler 可以获取到当前用户的身份信息来进行更细粒度的权限控制。

总结：构建可维护的微服务

认证和授权是任何严肃的后端服务都不可或缺的部分。通过 tower 的中间件模型，我们可以将这些复杂的逻辑从核心业务中优雅地剥离出来。

  1. tonic 拦截器：适用于简单的、无状态的、只读请求元数据的认证，如 API Key 检查。它是最简单快捷的认证方式。
  2. 自定义 tower::Layer: 提供了最大的灵活性和能力。适用于需要：
    • 维护状态（如限流器）。
    • 执行异步操作（如查询数据库）。
    • 修改响应。
    • 向后传递数据（通过请求 extensions）。
  3. 请求 extensions: 是在中间件和 handler 之间传递请求级别数据的标准方式。中间件负责“生产”数据（如 UserInfo），handler 负责“消费”它。
  4. 关注点分离: 通过将认证/授权逻辑移至中间件，我们的 gRPC handler (CrmService 的 impl) 可以更专注于核心业务逻辑。这使得代码更清晰，更易于单元测试（我们可以不经过中间件，直接测试 CrmService 的方法）。

这个“数据库驱动的认证 Layer + 业务逻辑 Service”的模式是一个非常通用且强大的架构。它不仅适用于 gRPC，同样适用于 axum。掌握了这个模式，你就掌握了构建可扩展、可维护、安全的 Rust 微服务的关键。

至此，我们的 CRM 微服务实战项目的主要部分已经完成。我们从建模开始，实现了核心业务逻辑，并为其添加了坚固的认证授权保护层。

思考题

  1. 在我们的 AuthService::call 方法中，我们执行了 self.inner.clone()。为什么每次调用都需要克隆内部服务？这和 Service::call 接收 &mut self 有什么关系？
  2. req.extensions_mut().insert(info) 插入的是 UserInfo 的所有权。在 handler 中，我们通过 request.extensions().get::<UserInfo>() 只能得到 &UserInfo。如果我们想在 handler 中获得 UserInfo 的所有权，该怎么做？Arc<UserInfo> 是一个好的选择吗？为什么？
  3. 我们的 AuthLayer 依赖于 PgPool。这意味着每次创建 AuthLayer 时都需要一个 PgPool 的克隆。请比较这种设计与直接在 AuthService::call 中从 AppState 中提取 PgPool 的优劣。
  4. 如果我们的认证逻辑非常复杂，包含多种认证方式（如 API Key, JWT, OAuth2），你会如何设计你的认证中间件？是创建一个巨大的 AuthLayer，还是将它们拆分为多个独立的 Layer 并组合起来？
  5. tonic 的 Status::unauthenticated 和 Status::permission_denied 有什么区别？请描述一个应该使用前者的场景和一个应该使用后者的场景。

实践练习

  1. 完善 list_contacts 的授权: 修改 list_contacts 方法。
    • 从 request.extensions() 中获取 UserInfo。
    • 如果用户是 admin，则允许他们通过请求中的 customer_id 查询任何客户的联系人。
    • 如果用户是普通用户，则忽略请求中的 customer_id，强制使用 user_info.customer_id 来查询，确保他们只能看到自己公司的联系人。
  2. 实现 JWT 认证：
    • 添加 jsonwebtoken 和 chrono crate。
    • 创建一个 /login 的 axum 路由，它接收用户名/密码，验证后生成一个 JWT token 并返回。
    • 修改你的 AuthLayer，使其不再检查 x-user-id，而是检查 Authorization: Bearer <jwt> 头部，验证 JWT 的签名和过期时间，并从中解析出 UserInfo。
  3. 创建一个 AdminOnly 中间件:
    • 编写一个新的 Layer，名为 AdminOnlyLayer。
    • 这个中间件应该在 AuthLayer 之后运行。
    • 它从 request.extensions() 中提取 UserInfo。
    • 如果用户的 roles 中不包含 "admin"，则直接返回 Status::permission_denied 错误。
    • 将这个 Layer 只应用到那些需要管理员权限的 gRPC 方法上（挑战：tonic 不像 axum 那样有 .route_layer()，你可能需要将服务拆分为 AdminCrmService 和 UserCrmService 并分别为它们应用不同的中间件栈）。
# 微服务 # 中间件 # 架构 # 开发语言 # rust
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区