rust-security

Rust 安全 - 快速参考

不要使用此技能的情况

• 一般 OWASP 概念 - 使用 owasp 或 owasp-top-10 技能
• Java 安全 - 使用 java-security 技能
• Python 安全 - 使用 python-security 技能
• 密钥管理 - 使用 secrets-management 技能

深入了解：使用 mcp__documentation__fetch_docs，并将技术指定为：rust，以获取 Rust 安全文档。

Rust 内置的安全优势

Rust 默认提供内存安全：

• 不会发生空指针解引用（改用 Option）
• 不会发生缓冲区溢出（边界检查）
• 不会发生释放后使用（所有权系统）
• 不会发生数据竞争（借用检查器）

不过，Rust 无法防护以下问题：

• 逻辑错误（授权缺陷）
• SQL 注入（字符串处理）
• XSS（模板处理）
• 密钥泄露
• 依赖项漏洞

依赖项审计

# cargo-audit - Check for known vulnerabilities
cargo install cargo-audit
cargo audit

# cargo-deny - Policy-based linting
cargo install cargo-deny
cargo deny check

# Check outdated dependencies
cargo install cargo-outdated
cargo outdated

# Snyk for Rust
snyk test

cargo-deny 配置（deny.toml）

[advisories]
vulnerability = "deny"
unmaintained = "warn"
yanked = "deny"

[licenses]
unlicensed = "deny"
allow = ["MIT", "Apache-2.0", "BSD-3-Clause"]

[bans]
multiple-versions = "warn"
wildcards = "deny"

[sources]
unknown-registry = "deny"
unknown-git = "deny"

CI/CD 集成

# GitHub Actions
- name: Security audit
  run: |
    cargo install cargo-audit
    cargo audit

- name: Dependency policy check
  run: |
    cargo install cargo-deny
    cargo deny check

防止 SQL 注入

SQLx - 安全（编译时检查）

use sqlx::{PgPool, query_as};

// SAFE - Compile-time verified query
let user: Option<User> = sqlx::query_as!(
    User,
    "SELECT * FROM users WHERE email = $1",
    email
)
.fetch_optional(&pool)
.await?;

// SAFE - Runtime query with bind
let user: Option<User> = sqlx::query_as::<_, User>(
    "SELECT * FROM users WHERE email = $1"
)
.bind(&email)
.fetch_optional(&pool)
.await?;

Diesel - 安全（类型安全的 ORM）

use diesel::prelude::*;

// SAFE - Type-safe query
let user = users::table
    .filter(users::email.eq(&email))
    .first::<User>(&mut conn)
    .optional()?;

// SAFE - Explicit parameter binding
diesel::sql_query("SELECT * FROM users WHERE email = $1")
    .bind::<Text, _>(&email)
    .load::<User>(&mut conn)?;

不安全的模式

// UNSAFE - String formatting
let query = format!("SELECT * FROM users WHERE email = '{}'", email);  // NEVER!

// UNSAFE - String concatenation
let query = "SELECT * FROM users WHERE email = '".to_owned() + &email + "'";  // NEVER!

防止 XSS

Askama（编译时模板 - 自动转义）

use askama::Template;

#[derive(Template)]
#[template(path = "page.html")]
struct PageTemplate<'a> {
    user_input: &'a str,  // Auto-escaped in template
}

<p>{{ user_input }}</p>


<p>{{ user_input|safe }}</p>  

Tera（运行时模板）

use tera::{Tera, Context};

let tera = Tera::new("templates/**/*")?;
let mut ctx = Context::new();
ctx.insert("user_input", &user_input);  // Auto-escaped

let rendered = tera.render("page.html", &ctx)?;

使用 ammonia 手动净化

use ammonia::clean;

// Sanitize HTML input
let safe_html = clean(&user_input);

// Custom policy
use ammonia::Builder;

let safe_html = Builder::default()
    .tags(hashset!["p", "b", "i", "a"])
    .url_schemes(hashset!["http", "https"])
    .link_rel(Some("noopener noreferrer"))
    .clean(&user_input)
    .to_string();

身份验证 - JWT

jsonwebtoken

use jsonwebtoken::{encode, decode, Header, Algorithm, Validation, EncodingKey, DecodingKey};
use serde::{Serialize, Deserialize};
use chrono::{Utc, Duration};

#[derive(Debug, Serialize, Deserialize)]
struct Claims {
    sub: String,       // user_id
    email: String,
    exp: usize,        // expiration
    iat: usize,        // issued at
}

fn generate_token(user_id: &str, email: &str, secret: &[u8]) -> Result<String, Error> {
    let expiration = Utc::now()
        .checked_add_signed(Duration::hours(1))
        .expect("valid timestamp")
        .timestamp() as usize;

    let claims = Claims {
        sub: user_id.to_owned(),
        email: email.to_owned(),
        exp: expiration,
        iat: Utc::now().timestamp() as usize,
    };

    encode(
        &Header::new(Algorithm::HS256),
        &claims,
        &EncodingKey::from_secret(secret)
    )
}

fn validate_token(token: &str, secret: &[u8]) -> Result<Claims, Error> {
    let mut validation = Validation::new(Algorithm::HS256);
    validation.validate_exp = true;

    let token_data = decode::<Claims>(
        token,
        &DecodingKey::from_secret(secret),
        &validation
    )?;

    Ok(token_data.claims)
}

使用 argon2 进行密码哈希

use argon2::{
    password_hash::{
        rand_core::OsRng,
        PasswordHash, PasswordHasher, PasswordVerifier, SaltString
    },
    Argon2
};

fn hash_password(password: &str) -> Result<String, Error> {
    let salt = SaltString::generate(&mut OsRng);
    let argon2 = Argon2::default();

    Ok(argon2
        .hash_password(password.as_bytes(), &salt)?
        .to_string())
}

fn verify_password(password: &str, hash: &str) -> Result<bool, Error> {
    let parsed_hash = PasswordHash::new(hash)?;
    Ok(Argon2::default()
        .verify_password(password.as_bytes(), &parsed_hash)
        .is_ok())
}

使用 validator 验证输入

use validator::{Validate, ValidationError};
use regex::Regex;
use lazy_static::lazy_static;

lazy_static! {
    static ref NAME_REGEX: Regex = Regex::new(r"^[a-zA-Z\s\-']+$").unwrap();
}

#[derive(Debug, Validate, Deserialize)]
struct CreateUserRequest {
    #[validate(email, length(max = 255))]
    email: String,

    #[validate(length(min = 12, max = 128), custom = "validate_password_strength")]
    password: String,

    #[validate(length(min = 2, max = 100), regex = "NAME_REGEX")]
    name: String,
}

fn validate_password_strength(password: &str) -> Result<(), ValidationError> {
    let has_upper = password.chars().any(|c| c.is_uppercase());
    let has_lower = password.chars().any(|c| c.is_lowercase());
    let has_digit = password.chars().any(|c| c.is_numeric());
    let has_special = password.chars().any(|c| "@$!%*?&".contains(c));

    if has_upper && has_lower && has_digit && has_special {
        Ok(())
    } else {
        Err(ValidationError::new("password_strength"))
    }
}

// Axum handler
async fn create_user(
    Json(payload): Json<CreateUserRequest>
) -> Result<Json<User>, AppError> {
    payload.validate()?;
    // payload is validated
}

安全文件上传（Axum）

use axum::{
    extract::Multipart,
    response::Json,
};
use tokio::fs::File;
use tokio::io::AsyncWriteExt;
use uuid::Uuid;

const MAX_FILE_SIZE: usize = 10 * 1024 * 1024; // 10 MB
const ALLOWED_TYPES: &[&str] = &["image/jpeg", "image/png", "application/pdf"];

async fn upload_file(mut multipart: Multipart) -> Result<Json<UploadResponse>, AppError> {
    while let Some(field) = multipart.next_field().await? {
        let content_type = field.content_type()
            .ok_or(AppError::BadRequest("Missing content type"))?;

        // Validate content type
        if !ALLOWED_TYPES.contains(&content_type) {
            return Err(AppError::BadRequest("File type not allowed"));
        }

        let data = field.bytes().await?;

        // Validate size
        if data.len() > MAX_FILE_SIZE {
            return Err(AppError::BadRequest("File too large"));
        }

        // Generate safe filename
        let ext = match content_type {
            "image/jpeg" => "jpg",
            "image/png" => "png",
            "application/pdf" => "pdf",
            _ => return Err(AppError::BadRequest("Unknown type")),
        };
        let safe_name = format!("{}.{}", Uuid::new_v4(), ext);

        // Save file
        let path = format!("uploads/{}", safe_name);
        let mut file = File::create(&path).await?;
        file.write_all(&data).await?;

        return Ok(Json(UploadResponse { filename: safe_name }));
    }

    Err(AppError::BadRequest("No file provided"))
}

CORS 配置（Axum）

use tower_http::cors::{CorsLayer, Any};
use http::{HeaderValue, Method};

let cors = CorsLayer::new()
    .allow_origin("https://myapp.com".parse::<HeaderValue>().unwrap())
    .allow_methods([Method::GET, Method::POST, Method::PUT, Method::DELETE])
    .allow_headers([http::header::AUTHORIZATION, http::header::CONTENT_TYPE])
    .allow_credentials(true);

let app = Router::new()
    .route("/api/users", get(get_users))
    .layer(cors);

安全标头中间件

use axum::{
    middleware::{self, Next},
    response::Response,
    http::Request,
};

async fn security_headers<B>(request: Request<B>, next: Next<B>) -> Response {
    let mut response = next.run(request).await;
    let headers = response.headers_mut();

    headers.insert("X-Content-Type-Options", "nosniff".parse().unwrap());
    headers.insert("X-Frame-Options", "DENY".parse().unwrap());
    headers.insert("X-XSS-Protection", "0".parse().unwrap());
    headers.insert("Referrer-Policy", "strict-origin-when-cross-origin".parse().unwrap());
    headers.insert("Content-Security-Policy", "default-src 'self'".parse().unwrap());
    headers.insert(
        "Strict-Transport-Security",
        "max-age=31536000; includeSubDomains".parse().unwrap()
    );

    response
}

// Apply to router
let app = Router::new()
    .route("/", get(index))
    .layer(middleware::from_fn(security_headers));

速率限制

use governor::{Quota, RateLimiter};
use nonzero_ext::nonzero;
use std::sync::Arc;

// Create rate limiter
let limiter = Arc::new(RateLimiter::direct(
    Quota::per_minute(nonzero!(10u32))
));

// Middleware
async fn rate_limit<B>(
    State(limiter): State<Arc<RateLimiter<...>>>,
    request: Request<B>,
    next: Next<B>
) -> Result<Response, StatusCode> {
    match limiter.check() {
        Ok(_) => Ok(next.run(request).await),
        Err(_) => Err(StatusCode::TOO_MANY_REQUESTS),
    }
}

密钥管理

use std::env;

#[derive(Clone)]
struct Config {
    jwt_secret: String,
    database_url: String,
    api_key: String,
}

impl Config {
    fn from_env() -> Result<Self, ConfigError> {
        Ok(Config {
            jwt_secret: env::var("JWT_SECRET")
                .map_err(|_| ConfigError::Missing("JWT_SECRET"))?,
            database_url: env::var("DATABASE_URL")
                .map_err(|_| ConfigError::Missing("DATABASE_URL"))?,
            api_key: env::var("API_KEY")
                .map_err(|_| ConfigError::Missing("API_KEY"))?,
        })
    }
}

// NEVER hardcode secrets
// const JWT_SECRET: &str = "hardcoded-secret";  // NEVER!

安全事件日志记录

use tracing::{info, warn};

fn log_login_attempt(username: &str, success: bool, ip: &str) {
    info!(
        user = username,
        success = success,
        ip = ip,
        "login attempt"
    );
}

fn log_access_denied(user_id: &str, resource: &str, ip: &str) {
    warn!(
        user_id = user_id,
        resource = resource,
        ip = ip,
        "access denied"
    );
}

// NEVER log sensitive data
// info!(password = password, "user data");  // NEVER!

不安全代码指南

// Minimize unsafe blocks
// Document why unsafe is necessary
// Encapsulate unsafe in safe abstractions

/// SAFETY: buffer is guaranteed to be valid UTF-8
/// because it was created from a valid String
unsafe fn process_buffer(buffer: &[u8]) -> &str {
    std::str::from_utf8_unchecked(buffer)
}

// Prefer safe alternatives
let s = std::str::from_utf8(buffer)?;  // Safe version

反模式

反模式	不良影响	正确做法
SQL 查询中使用 format!	SQL 注入	使用带绑定参数的查询宏
对用户输入使用 `	safe` 过滤器	XSS 漏洞
硬编码密钥	密钥泄露	使用环境变量
过多的 unsafe 代码块	绕过内存安全保护	尽量减少并记录 unsafe 代码
忽略 cargo audit 警告	已知漏洞	更新或替换依赖项
使用较弱的 JWT 算法	令牌伪造	至少使用 HS256
在处理程序中使用 unwrap()	生产环境中发生 panic	使用恰当的错误处理

快速故障排除

问题	可能原因	解决方案
cargo audit 发现 RUSTSEC	存在漏洞的 crate	使用 cargo update 更新
JWT 验证失败	算法或密钥错误	检查 Algorithm 枚举和密钥
CORS 错误	未配置来源	将来源添加到 CorsLayer
密码哈希速度慢	Argon2 参数过高	调整内存/迭代次数
SQLx 编译错误	查询与架构不匹配	运行 cargo sqlx prepare
模板未转义	使用了 `	safe` 过滤器

安全扫描命令

# Vulnerability audit
cargo audit

# Policy check
cargo deny check

# Clippy security lints
cargo clippy -- -W clippy::all -W clippy::pedantic

# Check for secrets
gitleaks detect
trufflehog git file://.

# SAST with semgrep
semgrep --config=p/rust .

相关技能

• OWASP Top 10:2025
• OWASP 通用指南
• 密钥管理
• 供应链安全