Tổng quan dự án

Trong bài học capstone này, bạn sẽ xây dựng một đại lý AI hoàn chỉnh thực hiện bốn tác vụ thực tế thay mặt một người dùng được xác thực: xác thực danh tính thông qua xác thực, đọc và ghi hồ sơ vào cơ sở dữ liệu, gọi các API bên ngoài để lấy hoặc gửi dữ liệu, và thực thi kiểm soát truy cập dựa trên vai trò (RBAC) để người dùng chỉ thấy những gì họ được phép xem. Đây là điểm kết nối nơi mọi khái niệm trước đó trong khóa học kết nối với nhau.

Đại lý bạn xây dựng theo mẫu yêu cầu-đáp lại: người dùng gửi một yêu cầu, đại lý kiểm tra người dùng là ai, quyết định người dùng được phép làm gì, sau đó truy vấn cơ sở dữ liệu, gọi API, hoặc cả hai, và trả về kết quả. Chúng ta sẽ dùng Python cho logic đại lý, cơ sở dữ liệu SQL cho sự tồn tại lâu dài, một API bên ngoài mô phỏng cho dữ liệu bên ngoài, và một cơ chế JWT (JSON Web Token) đơn giản cho xác thực.

Kiến trúc và Thiết lập

Trước khi viết mã, hãy phác thảo bốn phần động và cách chúng kết nối với nhau. Người dùng gửi yêu cầu kèm một mã thông báo JWT. Đại lý xác thực mã thông báo, trích xuất user ID và vai trò, rồi định tuyến yêu cầu đến bộ xử lý cơ sở dữ liệu hoặc người gọi API tùy thuộc vào ý định của người dùng. Bộ xử lý cơ sở dữ liệu sử dụng các truy vấn SQL có tham số để ngăn chặn tấn công chèn. Người gọi API gắn cùng JWT vào các yêu cầu ra ngoài để dịch vụ bên ngoài có thể xác thực người gọi độc lập.

from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: int
    role: str
    token: str

Đối tượng UserContext đi cùng với mọi cuộc gọi nội bộ. Khi một yêu cầu nhập vào đại lý, context được đánh dấu lên nó và mọi hàm hạ nguồn nhận được cùng một ngữ cảnh, do đó các kiểm tra cấp quyền xảy ra ở mọi bước, không chỉ ở điểm vào.

Triển khai Xác thực

Xác thực có nghĩa là chứng thực người dùng. Chúng ta sẽ sử dụng một token kiểu JWT đơn giản: một chuỗi được mã hóa base64 chứa user ID, vai trò và thời hạn hết hạn, được ký bằng một bí mật chia sẻ. Trong sản phẩm bạn sẽ dùng một thư viện như PyJWT, nhưng ở đây chúng ta giữ ở mức tối giản để bạn có thể nhìn thấy cơ chế.

import base64, json, hmac, hashlib, time

SECRET = "change-me-in-production"

def create_token(user_id: int, role: str) -> str:
    payload = {"uid": user_id, "role": role, "exp": int(time.time()) + 3600}
    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
    sig = hmac.new(SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
    return f"{body}.{sig}"

def verify_token(token: str) -> UserContext:
    body, sig = token.split(".")
    expected = hmac.new(SECRET.encode(), body.encode(), hashlib.sha256).hexdigest()
    if not hmac.compare_digest(sig, expected):
        raise PermissionError("Invalid token signature")
    payload = json.loads(base64.urlsafe_b64decode(body))
    if payload["exp"] < int(time.time()):
        raise PermissionError("Token expired")
    return UserContext(user_id=payload["uid"], role=payload["role"], token=token)

Chức năng verify_token là người canh cửa. Mọi điểm kết thúc được bảo vệ đều gọi nó trước. Nếu token bị giả mạo, hết hạn hoặc bị thiếu, yêu cầu bị từ chối trước khi bất kỳ mã cơ sở dữ liệu hoặc API được chạy.

Truy vấn Cơ sở dữ liệu một cách an toàn

Với xác thực được thực hiện xong, bước tiếp theo là cho phép tác nhân đọc hồ sơ người dùng. Quy tắc tối quan trọng ở đây là: không bao giờ xây dựng SQL bằng ghép chuỗi. Luôn sử dụng truy vấn có tham số để dữ liệu của người dùng được xử lý như dữ liệu, không phải SQL có thể thực thi.

import sqlite3

def get_user_orders(ctx: UserContext, target_user_id: int) -> list:
    conn = sqlite3.connect("app.db")
    cur = conn.cursor()
    cur.execute("SELECT id, product, total FROM orders WHERE user_id = ?", (target_user_id,))
    rows = cur.fetchall()
    conn.close()
    return rows

Quan sát cách ID người dùng được truyền như tham số dạng tuple, chứ không được nội suy vào chuỗi. Thói quen này duy nhất ngăn chặn tấn công SQL injection, loại tấn công phổ biến nhất nhắm vào cơ sở dữ liệu.

Gọi API bên ngoài

Nhiều tác vụ của tác nhân yêu cầu dữ liệu mà cơ sở dữ liệu không có, chẳng hạn trạng thái giao hàng từ một dịch vụ chuyển phát hoặc thời tiết cho tuyến đường giao hàng. Tác nhân gọi một API HTTP bên ngoài, truyền JWT tương tự để dịch vụ bên ngoài có thể xác minh người gọi.

import urllib.request

def get_shipping_status(ctx: UserContext, order_id: int) -> dict:
    req = urllib.request.Request(
        f"https://api.courier.example/orders/{order_id}",
        headers={"Authorization": f"Bearer {ctx.token}"}
    )
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())

Mã thông báo được chuyển tiếp, không được tạo mới. Điều này cho phép dịch vụ chuyển phát ghi lại ai đã yêu cầu và áp dụng giới hạn tốc độ theo từng người dùng.

Thực thi Kiểm soát Truy cập dựa trên Vai trò (RBAC)

Xác thực cho biết bạn là ai. Phân quyền (RBAC) cho biết họ được phép làm những gì. Nơi sạch nhất để thực thi điều này là một hàm policy nhỏ được gọi tại mọi điểm truy cập.

def require_role(ctx: UserContext, allowed: list[str]):
    if ctx.role not in allowed:
        raise PermissionError(f"Role '{ctx.role}' not permitted")

def list_all_orders(ctx: UserContext) -> list:
    require_role(ctx, ["admin"])
    conn = sqlite3.connect("app.db")
    cur = conn.cursor()
    cur.execute("SELECT id, user_id, product, total FROM orders")
    rows = cur.fetchall()
    conn.close()
    return rows

Một người dùng thường gọi list_all_orders sẽ bị từ chối. Một quản trị viên gọi nó sẽ thấy toàn bộ bảng. Kiểm tra policy chỉ mất một dòng, nhưng đó là sự khác biệt giữa một hệ thống an toàn và một vụ rò rỉ dữ liệu.

Liên kết tất cả lại với nhau

Chức năng tác nhân chính kết nối xác thực, truy cập cơ sở dữ liệu, gọi API và RBAC thành một luồng duy nhất.

def handle_request(token: str, intent: str, params: dict):
    ctx = verify_token(token)
    if intent == "my_orders":
        return get_user_orders(ctx, params["user_id"])
    elif intent == "all_orders":
        return list_all_orders(ctx)
    elif intent == "shipping":
        return get_shipping_status(ctx, params["order_id"])
    else:
        raise ValueError("Unknown intent")

Mọi đường dẫn bắt đầu bằng verify_token. Mọi đường dẫn chạm đến dữ liệu nhạy cảm gọi require_role. Mọi truy vấn cơ sở dữ liệu đều tham số hóa. Mọi cuộc gọi API ra ngoài mang theo token của người dùng. Bốn thói quen này, lặp lại khắp nơi, là những gì làm cho tác nhân sẵn sàng vận hành ở môi trường sản xuất.

Tóm tắt

Bạn hiện có một đại lý end-to-end hoạt động. Xác thực chứng minh danh tính qua các token đã ký. Các truy vấn SQL thao tác có tham số đọc và ghi dữ liệu mà không có rủi ro tiêm tấn công. Các cuộc gọi API bên ngoài tái sử dụng cùng một token để các dịch vụ bên ngoài có thể kiểm tra và giới hạn theo người dùng. Kiểm soát truy cập dựa trên vai trò được đặt tại mọi điểm vào, chứ không chỉ ở cửa ra vào, vì vậy việc leo thang đặc quyền sẽ đòi hỏi vượt qua nhiều kiểm tra. Khi bạn xây dựng các đại lý thực tế, mẫu nhận dạng vẫn giống nhau: xác thực một lần, phân quyền khắp nơi, tham số hóa luôn, và truyền tải bối cảnh qua từng cuộc gọi.