# Sinatra 安全技能 ## 第 1 层：快速参考 - 基本安全 ### CSRF 保护 # 启用机架::保护使用机架::保护 # 或者特别是 CSRF 使用 Rack::Protection::AuthenticityToken ### XSS预防 # 在 ERB 模板中 - 默认情况下总是转义 <%= user.bio %> # 转义（安全） <%== user.bio %> # 原始（危险！） # 在 JSON 响应中 - 使用正确的 JSON 编码需要“json” json({ 名称: 用户名 }.to_json) ### SQL注入预防 # 不好：字符串插值 DB["SELECT * FROM users WHERE email = '#{email}'"] # 好：参数化查询 DB["SELECT * FROM users WHERE email = ?", email] # 好：哈希条件用户.where（电子邮件：电子邮件） ### 安全会话使用机架::会话::Cookie， Secret: ENV['SESSION_SECRET'], # 长随机字符串同一站点：：严格， httponly：正确，安全：生产？ ### 输入验证帮手做的 def validate_email(电子邮件) email.to_s.match?(/\A[\w+\-.]+@[a-z\d\-]+(\.[a-z\d\-]+)*\.[a-z]+\z/i) 结束 def validate_integer(值) 整数（值）拯救参数错误、类型错误无结束结束发布'/用户'做暂停 400，“无效电子邮件”，除非 validate_email(params[:email]) # 进程... 结束 ### 身份验证检查帮手做的绝对认证！暂停 401, json({ error: '未经授权' }) 除非 current_user 结束 def 当前用户 @current_user ||= User.find_by(id: session[:user_id]) 结束结束在 '/admin/*' 之前验证！结束 ** ** ## 第 2 层：详细说明 - 安全实施 ### 全面的 CSRF 保护 **配置：** 类应用程序 < Sinatra::Base # 启用CSRF保护使用机架::保护::AuthenticityToken， except: [:json], # 跳过带有令牌身份验证的 JSON API 允许_if：->（环境）{ # 跳过具有不记名令牌的 API 端点 env['HTTP_AUTHORIZATION']&.start_with?('承载') } # 手动生成 CSRF 令牌帮手做的 def csrf_token 会话[:csrf] ||= SecureRandom.hex(32) 结束 def csrf_tag “<输入类型='隐藏'名称='authenticity_token'值='#{csrf_token}'>” 结束 def verify_csrf_token 令牌 = params[:authenticity_token] || request.env['HTTP_X_CSRF_TOKEN'] 暂停 403，“无效的 CSRF 令牌”，除非令牌 == session[:csrf] 结束结束 # 包含在表单中发布'/用户'做 verify_csrf_token 除非 request.content_type == 'application/json' # 进程... 结束结束 **浏览次数：** <表单方法=“POST”操作=“/用户”> <%= csrf_tag %> **For AJAX:** // 在 AJAX 请求中包含 CSRF 令牌获取（'/用户'，{ 方法：'POST'，标题：{ 'X-CSRF-Token': document.querySelector('[name=csrf_token]').value, '内容类型'：'应用程序/json' }, 正文：JSON.stringify(数据) }); ### XSS预防策略 **Template Escaping:** # ERB - 默认转义

<%= user_input %>

# 明确原始（仅适用于可信内容）

<%== trust_html %>

# Sanitize user HTML require 'sanitize' 帮手做的 def sanitize_html(html) Sanitize.fragment(html, Sanitize::Config::RELAXED) 结束结束 # In template

<%= sanitize_html(user_bio) %>

**JSON 响应：** # 始终使用正确的 JSON 编码获取 '/api/users/:id' 执行 user = User.find(params[:id]) # 不好：手动构建 JSON # "{ \"name\": \"#{user.name}\" }" # 如果名称包含引号，则 XSS # 好：使用 JSON 库内容类型：json { name: user.name, bio: user.bio }.to_json 结束 **内容安全政策：** 类应用程序 < Sinatra::Base 之前做标头“内容安全策略”=> [ “默认-src'自我'”， “script-src'self'https://cdn.example.com”， “style-src'self''不安全内联'”， “img-src‘自身’数据：https：”， “字体-src'自我'”， “连接-src'自我'”， “框架祖先‘无’” ].join(';') 结束结束 ### SQL注入预防 **参数化查询：** # 续集 # 不好 DB["从用户中选择 * WHERE name = '#{name}'"] # 好 DB["SELECT * FROM users WHERE name = ?", name] DB["SELECT * FROM users WHERE name = :name", name: name] # 活动记录 # 不好 User.where("电子邮件 = '#{email}'") # 好用户.where（电子邮件：电子邮件） User.where("电子邮件 = ?", 电子邮件) User.where("电子邮件 = :电子邮件", 电子邮件: 电子邮件) **输入验证：** 帮手做的 def validate_sql_param(参数, 类型: :string) 案例类型当：整数整数（参数）当：布尔值 [true, 'true', '1', 1].include?(param) 当：字符串 param.to_s.gsub(/['";\\]/, '') # 删除危险字符否则参数结束拯救参数错误停止 400，'无效参数' 结束结束获取 '/users/:id' 执行 id = validate_sql_param(params[:id], 类型: :integer) 用户=用户.find(id) json 用户.to_hash 结束 ### 身份验证模式 **密码验证：** 需要“bcrypt” 用户类包括BCrypt def 密码=(新密码) @password_hash = 密码.create(new_password) 结束 def 密码_散列 @密码哈希结束 def 验证（密码）密码.new(password_hash) == 密码结束结束 # 注册发布'/注册'做用户 = 用户.new( 电子邮件：参数[：电子邮件]，名称：参数[:名称] ）用户.密码 = 参数[:密码] 用户保存会话[:user_id] = user.id 重定向“/仪表板” 结束 # 登录发布'/登录'做 user = User.find_by(电子邮件: params[:电子邮件]) if user&.authenticate(params[:密码]) 会话[:user_id] = user.id 会话[:logged_in_at] = Time.now.to_i 重定向“/仪表板” 否则停止 401，“凭据无效” 结束结束 **基于令牌的身份验证：** 需要“jwt” TokenAuth 类秘密 = ENV['JWT_SECRET'] def self.encode(有效负载, exp = 24.hours.from_now) 负载[:exp] = exp.to_i JWT.encode（有效负载，秘密，'HS256'）结束 def self.decode(令牌) body = JWT.decode(token, SECRET, true, 算法: 'HS256')[0] HashWithIn DifferentAccess.new(body) 救援 JWT::DecodeError、JWT::ExpiredSignature 无结束结束 # 中间件 JWTAuth 类 def 初始化（应用程序） @应用=应用结束 def 调用(env) auth_header = env['HTTP_AUTHORIZATION'] token = auth_header&.split(' ')&.last 如果有效负载 = TokenAuth.decode(token) env['current_user_id'] = 负载[:user_id] @app.call(env) 否则 [401, { 'Content-Type' => 'application/json' }, ['{“错误”：“未经授权”}']] 结束结束结束 # 登录端点发布 '/api/login' 做 user = User.find_by(电子邮件: params[:电子邮件]) if user&.authenticate(params[:密码]) token = TokenAuth.encode(user_id: user.id) json({ 令牌: 令牌, 用户: user.to_hash }) 否则暂停 401, json({ 错误: '无效凭据' }) 结束结束 # 受保护的路由类 API < Sinatra::Base 使用 JWTAuth 帮手做的 def 当前用户 @current_user ||= User.find(request.env['current_user_id']) 结束结束获取 '/profile' 执行 json current_user.to_hash 结束结束 **API密钥认证：** APIKeyAuth 类 def 初始化（应用程序） @应用=应用结束 def 调用(env) api_key = env['HTTP_X_API_KEY'] 如果有效_api_key？（api_key）用户 = User.find_by(api_key: api_key) env['当前用户'] = 用户 @app.call(env) 否则 [401, { 'Content-Type' => 'application/json' }, ['{"error": "无效的 API 密钥"}']] 结束结束私人的 def valid_api_key?(key) key && User.exists?(api_key: key, active: true) 结束结束使用 APIKeyAuth # 生成API密钥帮手做的 def 生成_api_key 安全随机.hex(32) 结束结束发布“/api/keys” 验证！ api_key = 生成_api_key current_user.update(api_key: api_key) json({ api_key: api_key }) 结束 ### 授权模式 **基于角色的访问控制：** 用户类角色 = [:访客、:用户、:管理员、:超级管理员] def has_role?(角色) ROLES.index(self.role) >= ROLES.index(角色) 结束 def 可以吗？（动作，资源）案例角色当：管理员，：超级管理员真实当：用户行动==：阅读||资源.user_id == id 否则行动==：阅读结束结束结束帮手做的 def 授权！（操作，资源）除非 current_user&.can？（操作，资源）暂停 403, json({ 错误: '禁止' }) 结束结束结束 # 用法获取 '/posts/:id' 执行 post = Post.find(params[:id]) 授权！（：阅读、发布） json post.to_hash 结束删除 '/posts/:id' 做 post = Post.find(params[:id]) 授权！（：删除，发布）销毁后状态 204 结束 **基于权限的授权：** 类权限行动={ 帖子：[：创建、：读取、：更新、：删除]、用户：[：读取、：更新、：删除]、评论：[：创建、：读取、：删除] } def self.check(用户、操作、资源类型) 返回 false 除非用户权限=用户.权限权限.include?("#{resource_type}:#{action}") || 权限.include?("#{resource_type}:*") || 权限.include?("*:*") 结束结束帮手做的 def 可以吗？（操作，资源类型） Permission.check(当前用户、操作、资源类型) 结束 def 授权！（操作，资源类型）除非可以？（动作，资源类型）暂停 403, json({ 错误: '禁止' }) 结束结束结束发布 '/posts' 做授权！（：创建，：帖子） # 创建帖子结束 ### 速率限制 **使用机架::攻击:** 需要“机架/攻击” 类机架::攻击 # 限制登录尝试节流阀（'登录/ IP'，限制：5，周期：60）做 |req| req.ip if req.path == '/login' && req.post? 结束 # 通过 API 密钥限制 API 请求 throttle('api/key', 限制: 100, 周期: 60) do |req| req.env['HTTP_X_API_KEY'] if req.path.start_with?('/api') 结束 # 通过IP进行限制 throttle('req/ip', 限制: 300, 周期: 60) do |req| 请求IP 结束 # 阻止已知的不良行为者 blocklist('阻止不良 IP') do |req| BadIP.blocked?(req.ip) 结束 # 自定义响应 self.throttled_responder = lambda do |env| [ 429、 { '内容类型' => 'application/json' }, [{ error: '超出速率限制' }.to_json] ] 结束结束使用机架::攻击 ### 安全文件上传需要“安全随机” 文件上传处理类允许类型 = { '图像/jpeg' => '.jpg', '图像/png' => '.png', '图像/gif' => '.gif', '应用程序/pdf' => '.pdf' } 最大大小 = 5 * 1024 * 1024 # 5MB def self.process(文件) # 验证文件是否存在 return { error: '未提供文件' } 除非文件 # 验证文件大小如果文件[:临时文件].size > MAX_SIZE return { error: '文件太大' } 结束 # 验证内容类型内容类型 = 文件[:类型] 除非 ALLOWED_TYPES.key?(content_type) return { error: '无效的文件类型' } 结束 # 清理文件名原始名称 = File.basename(文件[:文件名]) sanitized_name = original_name.gsub(/[^a-zA-Z0-9\._-]/, '') # 生成唯一的文件名扩展名 = ALLOWED_TYPES[内容类型] unique_name = "#{SecureRandom.hex(16)}#{扩展名}" # 保存文件 upload_dir = '上传' FileUtils.mkdir_p（上传目录）路径 = File.join(upload_dir, unique_name) File.open(path, 'wb') 做 |f| f.write(文件[:临时文件].read) 结束 { 成功：true，路径：路径，文件名：unique_name } 结束结束发布'/上传'做结果 = FileUploadHandler.process(params[:file]) 如果结果[:错误] 暂停 400, json({ 错误: 结果[:错误] }) 否则 json({ url: "/uploads/#{结果[:文件名]}" }) 结束结束 ** ** ## 第 3 层：资源和示例 ### 安全标头 **综合安全标头：** 安全标头类标头= { 'X-Frame-Options' => '拒绝', 'X-Content-Type-Options' => 'nosniff', 'X-XSS-保护' => '1;模式=块', 'Referrer-Policy' => '跨源时严格源', '权限策略' => '地理位置=()、麦克风=()、摄像头=()', '严格传输安全' => '最大年龄=31536000;包含子域' } def 初始化（应用程序） @应用=应用结束 def 调用(env) 状态、标题、正文 = @app.call(env) headers.merge!(标题) [状态、标题、正文] 结束结束使用安全标头 ### OWASP 安全检查表请参阅“assets/owasp-checklist.md”以获取完整的清单，其中包括： 1. **预防注射** * SQL注入 * 命令注入 * LDAP注入 * XML注入 2. **身份验证失效** * 密码政策 * 会话管理 * 多重身份验证 * 账户锁定 3. **敏感数据暴露** * 静态加密 * 传输中加密 (HTTPS) * 安全密钥存储 * 数据最小化 4. **XML 外部实体 (XXE)** * XML解析器配置 * 禁用外部实体处理 5. **访问控制损坏** * 对所有受保护的路由进行身份验证 * 授权检查 * IDOR预防 * CORS配置 6. **安全配置错误** * 删除默认凭据 * 禁用目录列表 * 错误消息处理 * 保持依赖更新 7. **跨站脚本（XSS）** * 输出编码 * 输入验证 * 内容安全政策 * HTTPOnly cookie 8. **不安全的反序列化** * 验证序列化数据 * 使用安全的序列化格式 * 对序列化数据进行签名 9. **使用具有已知漏洞的组件** * 定期依赖更新 * 安全审计（捆绑审计） * 监控CVE数据库 10. **日志记录和监控不足** * 记录安全事件 * 监控攻击 * 警报系统 * 日志轮换和保留 ### 安全测试示例 **测试验证：** RSpec.describe '身份验证' 做描述“POST /login”做什么让（：用户）{创建（：用户，电子邮件：'test@example.com'，密码：'password123'）} 它“通过有效凭据成功” post '/login', { 电子邮件: 'test@example.com', 密码: 'password123' }.to_json, 'CONTENT_TYPE' => '应用程序/json' 期望（最后一个响应）.to be_ok 期望（json_response）.to have_key（'token'）结束它“因密码无效而失败” post '/login', { email: 'test@example.com', 密码: '错误' }.to_json, 'CONTENT_TYPE' => '应用程序/json' 期望（last_response.status）.to eq（401）结束它“防止暴力攻击” 6.次 post '/login', { email: 'test@example.com', 密码: '错误' }.to_json, 'CONTENT_TYPE' => '应用程序/json' 结束 Expect(last_response.status).to eq(429) # 速率限制结束结束结束 **测试授权：** RSpec.describe '授权' 做让（：用户）{创建（：用户）} 让（：管理员）{创建（：用户，角色：：管理员）} 让（：帖子）{创建（：帖子，用户：用户）} 描述 'DELETE /posts/:id' 做它“允许所有者删除” 删除“/posts/#{post.id}”、{}、auth_header(user.token) 期望（last_response.status）.to eq（204）结束它“允许管理员删除” 删除“/posts/#{post.id}”、{}、auth_header(admin.token) 期望（last_response.status）.to eq（204）结束它“拒绝其他用户”这样做 other_user = 创建(:用户) 删除“/posts/#{post.id}”、{}、auth_header(other_user.token) 期望（last_response.status）.to eq（403）结束它“需要身份验证” 删除“/posts/#{post.id}” 期望（last_response.status）.to eq（401）结束结束结束 ### 其他资源 * **安全中间件：** `assets/security-middleware.rb` * **身份验证模式：** `assets/auth-patterns.rb` * **OWASP 检查表：** `assets/owasp-checklist.md` * **安全审核模板：** `references/security-audit-template.md` * **渗透测试指南：** `references/penetration-testing.md`

西纳特拉安全

包含技能