[MCP学习笔记]MCP细粒度权限控制：动态 RBAC 实现

CREATE TABLE users (
user\_id INT PRIMARY KEY AUTO\_INCREMENT,
username VARCHAR(50) UNIQUE NOT NULL,
password\_hash VARCHAR(100) NOT NULL,
email VARCHAR(100),
created\_at TIMESTAMP DEFAULT CURRENT\_TIMESTAMP
);

CREATE TABLE roles (
role_id INT PRIMARY KEY AUTO_INCREMENT,
role_name VARCHAR(50) UNIQUE NOT NULL,
role_description TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE auth_policies (
policy_id INT PRIMARY KEY AUTO_INCREMENT,
role_id INT NOT NULL,
object_type VARCHAR(50) NOT NULL,
action_type VARCHAR(50) NOT NULL,
context_rules JSON NOT NULL, -- 存储上下文规则，如 {"time_range":"09:00-18:00","locations":["dept1","dept2"]}
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (role_id) REFERENCES roles(role_id)
);

from flask import Flask, request, jsonify
import mysql.connector

app = Flask(\\_\\_name\\_\\_)

# 数据库连接配置
db\\_config = {
    'user': 'root',
    'password': 'your\\_password',
    'host': 'localhost',
    'database': 'mcp\\_auth\\_db'
}

@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')

    if not username or not password:
        return jsonify({'error': 'Missing username or password'}), 400

    # 连接数据库验证用户
    try:
        conn = mysql.connector.connect(\\*\\*db\\_config)
        cursor = conn.cursor(dictionary=True)
        cursor.execute(
            "SELECT u.user\\_id, u.username, r.role\\_id, r.role\\_name FROM users u "
            "JOIN user\\_roles ur ON u.user\\_id = ur.user\\_id "
            "JOIN roles r ON ur.role\\_id = r.role\\_id "
            "WHERE u.username = %s AND u.password\\_hash = MD5(%s)",
            (username, password)
        )
        user\\_roles = cursor.fetchall()

        if not user\\_roles:
            return jsonify({'error': 'Invalid username or password'}), 401

        # 返回用户及其角色信息
        user\\_info = {
            'user\\_id': user\\_roles[0]['user\\_id'],
            'username': user\\_roles[0]['username'],
            'roles': [{'role\\_id': ur['role\\_id'], 'role\\_name': ur['role\\_name']} for ur in user\\_roles]
        }
        return jsonify(user\\_info), 200

    except mysql.connector.Error as err:
        print(f"Database error: {err}")
        return jsonify({'error': 'Internal server error'}), 500
    finally:
        if 'conn' in locals():
            conn.close()

from flask import request, g
import datetime
import geoip2.database  # 需提前安装 geoip2 库用于 IP 地理定位

# 加载 GeoIP 数据库，用于获取用户地理位置信息
geoip\\_db\\_path = 'GeoLite2-City.mmdb'  # 需提前下载并配置路径
geoip\\_reader = geoip2.database.Reader(geoip\\_db\\_path)

def collect\\_context():
    # 收集时间上下文
    current\\_time = datetime.datetime.now()
    g.context = {
        'time': {
            'current\\_time': current\\_time.isoformat(),
            'time\\_of\\_day': current\\_time.strftime('%H:%M'),
            'day\\_of\\_week': current\\_time.strftime('%A')
        },
        'client\\_ip': request.remote\\_addr
    }

    # 尝试获取地理位置上下文
    try:
        response = geoip\\_reader.city(g.context['client\\_ip'])
        g.context['location'] = {
            'country': response.country.name,
            'city': response.city.name,
            'latitude': response.location.latitude,
            'longitude': response.location.longitude
        }
    except:
        g.context['location'] = {
            'country': 'Unknown',
            'city': 'Unknown',
            'latitude': 0.0,
            'longitude': 0.0
        }

    # 可继续扩展收集设备类型、用户-agent 等更多上下文信息
    g.context['user\\_agent'] = request.headers.get('User-Agent')
    

import json
import mysql.connector
from flask import g

db\\_config = {
    'user': 'root',
    'password': 'your\\_password',
    'host': 'localhost',
    'database': 'mcp\\_auth\\_db'
}

def check\\_permission(user\\_roles, object\\_type, action\\_type):
    try:
        conn = mysql.connector.connect(\\*\\*db\\_config)
        cursor = conn.cursor(dictionary=True)

        # 查询匹配的角色权限策略
        query = """
            SELECT ap.context\\_rules 
            FROM auth\\_policies ap
            WHERE ap.role\\_id IN (%s) 
            AND ap.object\\_type = %s 
            AND ap.action\\_type = %s
        """
        # 构建 IN 子句的参数占位符
        role\\_ids\\_str = ','.join(['%s'] \\* len(user\\_roles))
        query = query % (role\\_ids\\_str, % s, % s)

        # 提取角色 ID 列表
        role\\_ids = [role['role\\_id'] for role in user\\_roles]

        # 执行查询
        cursor.execute(query, (\\*role\\_ids, object\\_type, action\\_type))
        policies = cursor.fetchall()

        if not policies:
            return False, "No matching policy found"

        # 遍历匹配策略，验证上下文规则
        for policy in policies:
            context\\_rules = json.loads(policy['context\\_rules'])

            # 验证时间规则（简化示例，仅验证时间范围）
            if 'time\\_range' in context\\_rules:
                current\\_time\\_str = g.context['time']['time\\_of\\_day']
                current\\_time = datetime.datetime.strptime(current\\_time\\_str, '%H:%M')
                time\\_range = context\\_rules['time\\_range'].split('-')
                start\\_time = datetime.datetime.strptime(time\\_range[0], '%H:%M')
                end\\_time = datetime.datetime.strptime(time\\_range[1], '%H:%M')

                if not (start\\_time <= current\\_time <= end\\_time):
                    continue  # 时间不匹配，跳过此策略

            # 验证位置规则（简化示例）
            if 'allowed\\_locations' in context\\_rules:
                user\\_city = g.context['location']['city']
                if user\\_city not in context\\_rules['allowed\\_locations']:
                    continue  # 位置不匹配，跳过此策略

            # 若通过所有验证规则，返回允许
            return True, "Permission granted by policy"

        # 所有策略验证未通过
        return False, "Permission denied by context rules"

    except mysql.connector.Error as err:
        print(f"Database error: {err}")
        return False, "Internal server error during permission check"
    finally:
        if 'conn' in locals():
            conn.close()
            

from flask import Flask, request, jsonify
from permission\\_checker import check\\_permission
from auth\\_service import login  # 引入登录接口
from context\\_collector import collect\\_context  # 引入上下文收集中间件

app = Flask(\\_\\_name\\_\\_)

# 注册中间件与登录接口
app.before\\_request(collect\\_context)
app.add\\_url\\_rule('/login', 'login', login, methods=['POST'])

@app.route('/resources/<resource\\_type>', methods=['GET', 'POST', 'DELETE'])
def access\\_resource(resource\\_type):
    # 获取用户认证信息（此处简化为从请求头获取 token，实际应结合更安全的认证机制）
    auth\\_token = request.headers.get('Authorization')
    if not auth\\_token:
        return jsonify({'error': 'Authentication required'}), 401

    # 模拟从 token 解析用户角色（实际应通过安全的 token 验证与解析机制）
    # 此处仅为示例，假设 token 包含 user\\_id 与 roles 信息
    try:
        token\\_payload = json.loads(auth\\_token)
        user\\_roles = token\\_payload.get('roles', [])
    except:
        return jsonify({'error': 'Invalid token'}), 401

    # 获取请求方法对应的操作类型
    action\\_map = {
        'GET': 'read',
        'POST': 'create',
        'DELETE': 'delete'
    }
    action\\_type = action\\_map.get(request.method, 'unknown')

    # 进行权限判定
    permission\\_granted, message = check\\_permission(user\\_roles, resource\\_type, action\\_type)

    if not permission\\_granted:
        return jsonify({'error': message}), 403

    # 权限通过，执行实际资源操作逻辑（此处仅为示例返回成功消息）
    return jsonify({'message': f'Successfully accessed {resource\\_type} resource', 'method': request.method}), 200

if \\_\\_name\\_\\_ == '\\_\\_main\\_\\_':
    app.run(host='0.0.0.0', port=5000, debug=True)
    

import unittest
from permission\\_checker import check\\_permission
from context\\_collector import collect\\_context
from flask import g

class TestPermissionChecker(unittest.TestCase):
    def setUp(self):
        # 模拟上下文信息
        g.context = {
            'time': {'time\\_of\\_day': '10:00'},
            'location': {'city': 'New York'}
        }

    def test\\_permission\\_granted(self):
        # 模拟用户角色
        user\\_roles = [{'role\\_id': 1, 'role\\_name': 'admin'}]
        object\\_type = 'documents'
        action\\_type = 'read'

        # 调用权限判定函数
        granted, message = check\\_permission(user\\_roles, object\\_type, action\\_type)

        self.assertTrue(granted)
        self.assertEqual(message, 'Permission granted by policy')

    def test\\_permission\\_denied\\_by\\_time(self):
        # 修改上下文时间，模拟时间不匹配场景
        g.context['time']['time\\_of\\_day'] = '20:00'

        user\\_roles = [{'role\\_id': 2, 'role\\_name': 'regular\\_user'}]
        object\\_type = 'sensitive\\_data'
        action\\_type = 'read'

        granted, message = check\\_permission(user\\_roles, object\\_type, action\\_type)

        self.assertFalse(granted)
        self.assertEqual(message, 'Permission denied by context rules')

if \\_\\_name\\_\\_ == '\\_\\_main\\_\\_':
    unittest.main()CREATE INDEX idx\\_auth\\_policies ON auth\\_policies (role\\_id, object\\_type, action\\_type);