Phân tích mã nguồn django-rest-framework 004: Ba cơ chế xác thực chính (Authentication/Permission/Throttle)

Tổng quan về ba module xác thực

Trong phương thức dispatch được override của APIView trong DRF, dòng lệnh self.initial(request, *args, **kwargs) chính là nơi thực thi ba cơ chế xác thực. Khi xem xét chi tiết, bạn sẽ thấy thứ tự thực hiện lần lượt là: xác thực (authentication), phân quyền (permission), và giới hạn tốc độ (throttle).


# Đảm bảo request đến được phép
self.perform_authentication(request)
self.check_permissions(request)
self.check_throttles(request)

Nói một cách đơn giản, xác thực có nhiệm vụ kiểm tra thông tin người dùng được gửi kèm trong request.

Nếu thông tin người dùng hợp lệ, tiến hành kiểm tra phân quyền, tức là xác định xem người dùng này có quyền truy cập trang hay không.

Nếu cũng có quyền truy cập, thì tiến hành kiểm tra giới hạn tốc độ (tần suất), tức là xem người dùng này có truy cập trang quá thường xuyên hay không (đây là một chiến lược chống thu thập dữ liệu tự động).

Module xác thực (Authentication)

Một số trang yêu cầu người dùng phải đăng nhập mới có thể truy cập. Vì HTTP là giao thức không trạng thái, không thể trực tiếp mang trạng thái đăng nhập của người dùng trong request. Do đó, phương pháp phổ biến để xác định xem người dùng đã đăng nhập hay chưa là: sau khi người dùng đăng nhập, hệ thống backend tạo và ghi lại một token, sau đó trả token này về trình duyệt. Lần sau khi truy cập trang cần đăng nhập, request sẽ mang theo token này. Backend nhận được token, sau khi so sánh hoặc giải mã, lấy được thông tin người dùng tương ứng, nếu hợp lệ thì cho phép truy cập trang hoặc tiếp tục các bước kiểm tra tiếp theo.

Phân tích luồng mã nguồn

Nhấp vào hàm entry của module xác thực self.perform_authentication(request), bạn sẽ thấy bên trong chỉ có một dòng đơn giản request.user. Điều này giống như việc lấy thuộc tính user của request. Như đã biết từ chương trước, request này là một request do DRF đóng gói. Vì vậy, chúng ta đến với lớp rest_framework.request.Request.


@property
def user(self):
    """
    Trả về người dùng được liên kết với request hiện tại,
    được xác thực bởi các lớp xác thực được cung cấp cho request.
    """
    if not hasattr(self, '_user'):
        with wrap_attributeerrors():
            self._authenticate()
    return self._user

Đây là một phương thức được trang trí bằng property (thuộc tính), do đó chỉ cần gọi .user là nó sẽ thực thi. Phương thức này trả về thuộc tính _user của self. Trước đó chúng ta chưa gán giá trị cho thuộc tính này. Theo điều kiện if ở trên, có thể thấy ban đầu request không có thuộc tính _user, điều đó có nghĩa là trong self._authenticate() phải có logic gán giá trị.

Nhắc lại, trong hàm self.perform_authentication(request) ở trên chỉ đơn giản gọi request.user mà không trả về giá trị nào, điều này cho thấy self.perform_authentication(request) không cần trả về gì, mà chỉ muốn chương trình thực thi phương thức self._authenticate().

Nhấp vào _authenticate():


def _authenticate(self):
    """
    Cố gắng xác thực request bằng cách sử dụng lần lượt từng instance xác thực.
    """
    # Duyệt qua danh sách authenticator
    for authenticator in self.authenticators:
        try:
            # Thực thi phương thức authenticate của authenticator, trả về kết quả xác thực dưới dạng tuple (user, auth)
            user_auth_tuple = authenticator.authenticate(self)
        except exceptions.APIException:
            # Nếu phương thức authenticate của authenticator ném ra ngoại lệ APIException,
            # điều đó có nghĩa là logic xác thực có thể đã gặp lỗi
            # thì thực thi phương thức _not_authenticated
            self._not_authenticated()
            # Sau khi thực thi, tiếp tục ném ngoại lệ ra ngoài
            raise

        # Nếu phương thức authenticate của authenticator trả về tuple không rỗng,
        # thì gán ba thuộc tính cho request: _authenticator, user, auth
        if user_auth_tuple is not None:
            self._authenticator = authenticator
            self.user, self.auth = user_auth_tuple
            return

    # Nếu không, tiếp tục thực thi phương thức _not_authenticated
    self._not_authenticated()

def _not_authenticated(self):
    """
    Đặt authenticator, user & authtoken để đại diện cho một request chưa được xác thực.

    Mặc định là None, AnonymousUser & None.
    """
    # Phương thức này cho thấy request không có thông tin xác thực, sẽ đặt:
    # Thuộc tính _authenticator là None,
    # Thuộc tính user là giá trị của api_settings.UNAUTHENTICATED_USER(), giá trị này có thể được tùy chỉnh trong file settings,
    #      Giá trị mặc định là đối tượng người dùng ẩn danh AnonymousUser do Django định nghĩa
    # Thuộc tính auth là giá trị của api_settings.UNAUTHENTICATED_TOKEN(), giá trị này cũng có thể được tùy chỉnh trong file settings,
    #      Giá trị mặc định là None
    self._authenticator = None

    if api_settings.UNAUTHENTICATED_USER:
        self.user = api_settings.UNAUTHENTICATED_USER()
    else:
        self.user = None

    if api_settings.UNAUTHENTICATED_TOKEN:
        self.auth = api_settings.UNAUTHENTICATED_TOKEN()
    else:
        self.auth = None

Có thể thấy, phương thức _authenticate() có thể trả về hai kết quả:

Kết quả thứ nhất: Khi duyệt đến một authenticator nào đó, phương thức authenticate(self) của nó trả về một tuple (user, auth) không rỗng, nghĩa là xác thực thành công.
Kết quả thứ hai: Khi authenticator chạy lỗi hoặc trả về tuple rỗng, nghĩa là request không mang thông tin xác thực, mặc định trả về người dùng ẩn danh của Django. Giá trị trả về này có thể được tùy chỉnh trong settings của chúng ta.

Điều này cho thấy logic xác thực chi tiết vẫn nằm trong phương thức authenticate() của authenticator. Theo kinh nghiệm trước đây, các loại 'trình' (parser, renderer, v.v.) trong DRF thường được tạo thành một danh sách các đối tượng, sau đó gọi hàm của từng đối tượng trong một vòng lặp. Authenticator cũng vậy. Đoạn mã trên cho thấy request có một thuộc tính authenticators, thuộc tính này phải là một danh sách các authenticator. Tuy nhiên, trong quá trình phân tích, chúng ta không thấy logic gán giá trị cho thuộc tính này. Hãy nhớ lại rằng logic gán này thực sự được thực hiện ở bước trước ba bước xác thực của dispatch, khi khởi tạo đối tượng Request.


def get_authenticators(self):
    """
    Khởi tạo và trả về danh sách các authenticator mà view này có thể sử dụng.
    """
    return [auth() for auth in self.authentication_classes]

Nhấp vào authentication_classes, có thể thấy DRF có hai authenticator mặc định: SessionAuthentication và BasicAuthentication.


'DEFAULT_AUTHENTICATION_CLASSES': [
    'rest_framework.authentication.SessionAuthentication',
    'rest_framework.authentication.BasicAuthentication'
]

Điều này có nghĩa là mặc định, mỗi request DRF đều sẽ đi qua hai authenticator này. Vậy tại sao không có lỗi xác thực? Rõ ràng chúng ta không cung cấp thông tin xác thực. Hãy xem xét mã nguồn của BasicAuthentication để tìm hiểu.

BasicAuthentication

Đến với rest_framework.authentication.py, có thể thấy DRF định nghĩa một số authenticator có sẵn, như BasicAuthentication, RemoteUserAuthentication, SessionAuthentication, TokenAuthentication. Chúng ta phân tích mã nguồn của BasicAuthentication để hiểu rõ hơn và có thể tự viết authenticator tùy chỉnh.

Quy tắc xác thực

Trước tiên, hãy nói về quy tắc xác thực. Để xác thực, cần có một quy tắc xác thực được định nghĩa trước để xác định xem xác thực có thành công hay không. Quy tắc xác thực BasicAuth là: thêm một chuỗi vào tham số Authorization của header request, chuỗi này có định dạng là 'basic + khoảng trắng + secret'. basic là tiền tố cố định, có thể coi là 'muối' (salt). Secret là một chuỗi được tạo ra bằng cách mã hóa tên người dùng và mật khẩu.

Khi xác thực, trước tiên lấy giá trị của Authorization, kiểm tra xem định dạng có đúng quy tắc hay không. Nếu không đúng, trả về None. Nếu đúng, giải mã secret để lấy tên người dùng và mật khẩu, sau đó so sánh với tên người dùng và mật khẩu trong cơ sở dữ liệu. Nếu không khớp, ném ra ngoại lệ; nếu khớp, quá trình xác thực kết thúc thành công.

Phân tích mã nguồn

Như đã biết, phương thức cốt lõi của authenticator là authenticate(), phương thức này bắt buộc phải có khi tạo authenticator tùy chỉnh.


class BasicAuthentication(BaseAuthentication):
    """
    Xác thực HTTP Basic dựa trên tên người dùng và mật khẩu.
    """
    www_authenticate_realm = 'api'

    def authenticate(self, request):
        """
        Trả về một đối tượng `User` nếu tên người dùng và mật khẩu chính xác
        được cung cấp qua HTTP Basic Authentication. Nếu không, trả về `None`.
        """
        # Tách giá trị của tham số Authorization trong header request bằng dấu cách thành một list
        auth = get_authorization_header(request).split()

        # Nếu list rỗng hoặc phần tử đầu tiên không phải 'basic', trả về None
        if not auth or auth[0].lower() != b'basic':
            return None

        # Nếu độ dài auth không bằng 2, ném ra ngoại lệ
        if len(auth) == 1:
            msg = _('Invalid basic header. No credentials provided.')
            raise exceptions.AuthenticationFailed(msg)
        elif len(auth) > 2:
            msg = _('Invalid basic header. Credentials string should not contain spaces.')
            raise exceptions.AuthenticationFailed(msg)

        # Giải mã để lấy tên người dùng và mật khẩu
        try:
            auth_parts = base64.b64decode(auth[1]).decode(HTTP_HEADER_ENCODING).partition(':')
        except (TypeError, UnicodeDecodeError, binascii.Error):
            msg = _('Invalid basic header. Credentials not correctly base64 encoded.')
            raise exceptions.AuthenticationFailed(msg)

        userid, password = auth_parts[0], auth_parts[2]
        # Xác thực tên người dùng và mật khẩu đã lấy được
        return self.authenticate_credentials(userid, password, request)

    def get_authorization_header(request):
        """
        Trả về header 'Authorization:' của request dưới dạng bytestring.

        Giấu đi một số hành vi kỳ lạ của test client khi header có thể là unicode.
        """
        # Lấy giá trị tham số Authorization trong header request
        #    Trong DRF, vì request đã được đóng gói, có thể sử dụng request._request.headers.get()
        #    hoặc request.META.get(), nhưng META đã biến đổi tham số gốc:
        #      a) Đổi dấu gạch ngang (-) thành dấu gạch dưới (_)
        #      b) Chuyển tất cả chữ cái thành chữ hoa
        #      c) Thêm tiền tố HTTP_ vào trước tham số gốc
        auth = request.META.get('HTTP_AUTHORIZATION', b'')
        # Mã hóa auth thành iso-8859-1
        if isinstance(auth, str):
            # Xử lý hành vi kỳ lạ của test client trong Django
            auth = auth.encode(HTTP_HEADER_ENCODING)
        return auth

    def authenticate_credentials(self, userid, password, request=None):
        """
        Xác thực userid và password với tên người dùng và mật khẩu,
        có thể kèm request để lấy context.
        """
        credentials = {
            get_user_model().USERNAME_FIELD: userid,
            'password': password
        }
        # Xác thực mật khẩu người dùng, lấy đối tượng User
        user = authenticate(request=request, **credentials)

        # Nếu đối tượng user rỗng, ném ra ngoại lệ
        if user is None:
            raise exceptions.AuthenticationFailed(_('Invalid username/password.'))

        # Nếu đối tượng user chưa được kích hoạt, ném ra ngoại lệ
        if not user.is_active:
            raise exceptions.AuthenticationFailed(_('User inactive or deleted.'))

        # Lấy được đối tượng User thành công, trả về tuple
        return (user, None)

Qua mã nguồn, có thể thấy có ba kết quả trả về:

None: Khi giá trị đầu tiên của tham số Authorization không phải là 'basic'. Kết hợp với luồng bên ngoài _authenticate(), cuối cùng sẽ coi như không có thông tin xác thực và trả về người dùng ẩn danh AnonymousUser.
raise exceptions.AuthenticationFailed: Được ném ra trong hầu hết các trường hợp xác thực thất bại. Luồng bên ngoài _authenticate() không bắt ngoại lệ này, nó tiếp tục được ném ra ngoài và cuối cùng bị bắt trong dispatch, được chuyển cho module xử lý ngoại lệ để trả về kết quả.
return (user, None): Được trả về khi xác thực thành công. Phải trả về một tuple. Kết hợp với luồng bên ngoài _authenticate(), nó sẽ gán ba thuộc tính cho request.

Tạo lớp xác thực tùy chỉnh

Chúng ta có thể bắt chước BasicAuthentication để tạo một lớp xác thực tùy chỉnh gọi là MyAuthentication. Quy tắc là: thực hiện xác thực BasicAuthentication trước, nếu không thành công, có thể thực hiện xác thực MyAuthentication. Quy tắc xác thực của chúng ta là: truyền một tham số my-token trong header request, định dạng chuỗi là 'auth_sort/username'. Nếu định dạng đúng và username tồn tại trong cơ sở dữ liệu, xác thực thành công. Ví dụ mã:


class MyAuthentication(BaseAuthentication):
    def authenticate(self, request):
        """
        Quy ước: mang tham số my-token trong header request, định dạng auth_sort/username
        Lưu ý:
            1. Trong header HTTP, tốt nhất không nên dùng dấu gạch dưới, vì có thể không lấy được tham số này.
            2. Trong DRF, vì request đã được đóng gói, có thể sử dụng request._request.headers.get()
               hoặc request.META.get(), nhưng META đã biến đổi tham số gốc:
                    a) Đổi dấu gạch ngang (-) thành dấu gạch dưới (_)
                    b) Chuyển tất cả chữ cái thành chữ hoa
                    c) Thêm tiền tố HTTP_ vào trước tham số gốc
                Vậy tham số gốc my-token sẽ trở thành HTTP_MY_TOKEN
        """
        # Lấy tham số token
        token = request.META.get('HTTP_MY_TOKEN')
        # print(request._request.headers.get('my-token'))
        if not token:
            raise exceptions.AuthenticationFailed('Không có token, xác thực thất bại')
        # Tách token
        auth = token.split('/')
        if len(auth) != 2 or auth[0] != 'auth_sort':
            raise exceptions.AuthenticationFailed('Sai định dạng token')
        # Lấy tên người dùng và kiểm tra xem người dùng có tồn tại không
        user = User.objects.filter(username=auth[1])
        if not user:
            raise exceptions.AuthenticationFailed('Tên người dùng không tồn tại')
        # Xác thực thành công
        return user, None


class BookAuthViewSet(mixins.ListModelMixin, GenericViewSet):
    queryset = Book.objects.filter(is_delete=False)
    serializer_class = BookSerializer
    authentication_classes = [BasicAuthentication, MyAuthentication, ]

    def list(self, request, *args, **kwargs):
        print(request.user)
        response = super().list(request, *args, **kwargs)
        return MyResponse(response.data)

Module phân quyền (Permission)

Cách phân tích mã tương tự module xác thực. Nhấp vào self.check_permissions(request), bạn sẽ thấy self.get_permissions() trả về một danh sách các permission class. Vòng lặp gọi phương thức has_permission(request, self) của từng permission class để kiểm tra xem có quyền truy cập hay không. Nếu không có quyền, ném ra ngoại lệ raise exceptions.PermissionDenied(detail=message). Đến rest_framework.settings để xem lớp permission mặc định là AllowAny.


'DEFAULT_PERMISSION_CLASSES': [
    'rest_framework.permissions.AllowAny',
]

Đến lớp rest_framework.permissions.AllowAny, có thể thấy phương thức has_permission của nó chỉ đơn giản trả về True, nghĩa là đúng như tên gọi, bất kỳ request nào cũng được phép truy cập.


class AllowAny(BasePermission):
    """
    Cho phép mọi quyền truy cập.
    Điều này không bắt buộc, vì bạn có thể sử dụng danh sách permission_classes rỗng,
    nhưng nó hữu ích vì nó làm rõ ý định hơn.
    """

    def has_permission(self, request, view):
        return True

Cần lưu ý rằng ngoài has_permission(self, request, view), lớp permission còn có thể định nghĩa has_object_permission(self, request, view, obj). Phương thức xác thực quyền này dành riêng cho một đối tượng model cụ thể, trong khi has_permission là giới hạn quyền cho lớp view.

DRF cũng định nghĩa các lớp permission khác, như IsAdminUser (user.is_staff phải là True), IsAuthenticated (người dùng đã qua xác thực), IsAuthenticatedOrReadOnly (phương thức request là GET/HEAD/OPTIONS hoặc người dùng đã qua xác thực).

Nếu xác thực quyền thất bại, thông báo lỗi có thể được định nghĩa trong thuộc tính message của lớp permission. Trong phương thức check_permissions(self, request) của dispatch, nếu permission.has_permission trả về False, thì self.permission_denied(request, message=getattr(permission, 'message', None)) sẽ lấy thuộc tính message của permission class làm thông báo lỗi.

Tạo lớp permission tùy chỉnh

Để tạo permission tùy chỉnh, cần kế thừa lớp BasePermission và triển khai một hoặc cả hai phương thức sau:

has_permission(self, request, view)
has_object_permission(self, request, view, obj)

Nếu request được cấp quyền truy cập, phương thức sẽ trả về True, nếu không thì trả về False.

Ở đây chúng ta định nghĩa một quy tắc permission như sau: người dùng ẩn danh không thể xem dữ liệu book; người dùng thông thường chỉ có thể xem thông tin chi tiết của những cuốn sách do nhà xuất bản có id là 1 phát hành; người dùng quản trị viên có thể xem tất cả dữ liệu.


class MyPermission(BasePermission):
    """
    Người dùng ẩn danh không thể xem dữ liệu book,
    người dùng thông thường chỉ có thể xem thông tin chi tiết của sách do nhà xuất bản id=1 phát hành,
    người dùng quản trị viên có thể xem tất cả dữ liệu.
    """
    message = 'Người dùng không phải quản trị viên chỉ có thể xem dữ liệu của nhà xuất bản Thượng Hải'

    def has_permission(self, request, view):
        # Người dùng ẩn danh trả về False
        return not isinstance(request.user, AnonymousUser)

    def has_object_permission(self, request, view, obj):
        # Quản trị viên hoặc người dùng không phải quản trị viên có publish_id của sách là 1, trả về True
        return bool(request.user.is_staff or (request.user.is_staff is False and obj.publish_id == 1))


class BookAuthViewSet(mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet):
    queryset = Book.objects.filter(is_delete=False)
    serializer_class = BookSerializer
    authentication_classes = [BasicAuthentication, MyAuthentication]
    permission_classes = [MyPermission, ]

    def list(self, request, *args, **kwargs):
        print(request.user)
        response = super().list(request, *args, **kwargs)
        return MyResponse(result=response.data)

Module giới hạn tốc độ (Throttle)

Giới hạn tốc độ (throttle) có thể được hình dung như một van tiết lưu, chỉ ra một trạng thái tạm thời để kiểm soát số lần client được phép gửi request đến API trong một khoảng thời gian nhất định, tức là tần suất. Ví dụ, bạn có thể muốn giới hạn người dùng tối đa 60 request mỗi phút, 1000 request mỗi ngày.

Cách phân tích mã tương tự module xác thực. Nhấp vào self.check_throttles(request), bạn sẽ thấy self.get_throttles() trả về một danh sách các throttle class. Vòng lặp gọi phương thức allow_request(request, self) của từng throttle class để kiểm tra xem có được phép truy cập hay không. Nếu không được phép, gọi phương thức wait() của throttle class để lấy thời gian cần chờ, lưu trữ thời gian chờ của từng throttle class trong mảng throttle_durations, tìm thời gian lớn nhất, và ném ra ngoại lệ raise exceptions.Throttled(wait).


def check_throttles(self, request):
    """
    Kiểm tra xem request có nên bị giới hạn tốc độ hay không.
    Ném ra ngoại lệ thích hợp nếu request bị giới hạn tốc độ.
    """
    throttle_durations = []
    for throttle in self.get_throttles():
        if not throttle.allow_request(request, self):
            throttle_durations.append(throttle.wait())

    if throttle_durations:
        # Lọc ra các giá trị `None` có thể xảy ra trong trường hợp thay đổi cấu hình / tỷ lệ
        durations = [
            duration for duration in throttle_durations
            if duration is not None
        ]

        duration = max(durations, default=None)
        self.throttled(request, duration)

Đến settings, có thể thấy danh sách throttle mặc định là rỗng, nghĩa là không giới hạn tốc độ.

Đến rest_framework.throttling.py, có thể thấy một số throttle class đã được định nghĩa, như SimpleRateThrottle(BaseThrottle), AnonRateThrottle(SimpleRateThrottle), UserRateThrottle(SimpleRateThrottle), ScopedRateThrottle(SimpleRateThrottle).

Qua quan hệ kế thừa, có thể thấy SimpleRateThrottle là lớp cha của các throttle class khác. Các phương thức bắt buộc phải có của throttle class là allow_request() và wait() đều được định nghĩa trong lớp SimpleRateThrottle. Các lớp con chỉ ghi đè phương thức get_cache_key(self, request, view). Hãy xem xét mã nguồn của SimpleRateThrottle để hiểu rõ hơn và có thể tự viết throttle class tùy chỉnh.

Phân tích mã nguồn SimpleRateThrottle

Theo docstring của lớp, đây là một bộ giới hạn tần suất đơn giản được triển khai thông qua cache. Tần suất cần được đặt ở định dạng 'số lần request / chu kỳ', ví dụ: 3 lần mỗi giây là '3/s'. Ở đây chúng ta lấy ví dụ là 3 lần mỗi phút ('3/m') để giải thích mã.

Đi thẳng vào phương thức allow_request(), đây là điểm vào của chương trình.


def allow_request(self, request, view):
    """
    Triển khai kiểm tra xem request có nên bị giới hạn tốc độ hay không.

    Nếu thành công, gọi `throttle_success`.
    Nếu thất bại, gọi `throttle_failure`.
    """
    # Nếu chưa đặt tần suất, cho phép request đi qua
    if self.rate is None:
        return True

    # Lấy key của cache, nếu key rỗng, cho phép request đi qua
    self.key = self.get_cache_key(request, view)
    if self.key is None:
        return True

    # Lấy giá trị cache tương ứng với key
    self.history = self.cache.get(self.key, [])
    # Ghi lại thời gian hiện tại
    self.now = self.timer()

    # Xóa các request trong history đã hết thời hạn (đã qua khoảng thời gian throttle)
    while self.history and self.history[-1] <= self.now - self.duration:
        self.history.pop()
    # Nếu độ dài history lớn hơn hoặc bằng số request giới hạn, nghĩa là số request đã vượt quá, không qua được kiểm tra tần suất
    if len(self.history) >= self.num_requests:
        return self.throttle_failure()
    # Nếu không, đã qua kiểm tra tần suất, cho phép request
    return self.throttle_success()

Chúng ta đã lấy được giá trị của self.key, bước tiếp theo là lấy giá trị tương ứng với key từ cache và gán cho thuộc tính history. Nếu không lấy được, trả về một danh sách rỗng. Điều này cho thấy history có lẽ là một danh sách ghi lại lịch sử. Phân tích ở đây mô phỏng bước của request đầu tiên, trong các bước trước đó chưa có logic chèn dữ liệu vào cache, vì vậy history nhận được sẽ là một danh sách rỗng.

Vì là danh sách rỗng, sẽ không vào vòng lặp while, độ dài của history cũng sẽ không >= self.num_requests, trừ khi self.num_requests cũng bằng 0. Có thể thấy self.duration và self.num_requests chưa được gán giá trị trước đó. Nhấp vào để xem hai thuộc tính này là gì.

Hai tham số này được gán giá trị trong quá trình __init__ (khởi tạo): self.num_requests, self.duration = self.parse_rate(self.rate).

Trước tiên, xem thuộc tính self.rate được lấy như thế nào. Nó được lấy thông qua get_rate(self), thực chất là một thuộc tính lớp THROTTLE_RATES. Thuộc tính lớp này có thể được định nghĩa trong throttle class tùy chỉnh (local) hoặc trong settings của dự án (global). Định nghĩa mặc định là:


# Throttling
'DEFAULT_THROTTLE_RATES': {
    'user': None,
    'anon': None,
}

Quay lại phương thức parse_rate(self.rate):


def parse_rate(self, rate):
    """
    Với chuỗi tỷ lệ request, trả về một tuple gồm:
    , 
    """
    if rate is None:
        return (None, None)
    num, period = rate.split('/')
    num_requests = int(num)
    duration = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}[period[0]]
    return (num_requests, duration)

Phương thức này tách rate bằng dấu '/' và phân tích cú pháp. Ví dụ: rate='3/m' có nghĩa là tối đa 3 lần truy cập mỗi phút, do đó num_requests=3, duration=60, nghĩa là cuối cùng chuyển đổi thành tối đa xxx lần truy cập trong xxx giây. Nếu num_requests=0, nghĩa là không được phép truy cập lần nào, tất cả đều bị từ chối.

Sau khi hiểu rõ hai thuộc tính này, quay lại allow_request(), thấy rằng nếu num_requests không bằng 0, request đầu tiên chắc chắn sẽ thành công. Khi đó, chúng ta đi đến phương thức throttle_success().


def throttle_success(self):
    """
    Chèn dấu thời gian của request hiện tại cùng với key vào cache.
    """
    # Chèn thời gian hiện tại vào đầu danh sách history
    self.history.insert(0, self.now)
    # Đặt một bản ghi vào cache, key là key do get_cache_key trả về, value là danh sách history, thời gian hết hạn là xxx giây do rate quy định
    self.cache.set(self.key, self.history, self.duration)
    return True

Phương thức này làm hai việc: thứ nhất, chèn thời gian hiện tại vào đầu danh sách history; thứ hai, chèn danh sách history vào cache và đặt thời gian hết hạn. Ví dụ: lúc này history=['2020-07-23 10:30:00'].

Với ví dụ tối đa 3 lần truy cập mỗi phút:

Khi truy cập lần thứ hai trong vòng một phút (ví dụ lúc '2020-07-23 10:30:10'): len(history) = 1, kiểm tra vẫn thành công. Lúc này history=['2020-07-23 10:30:10', '2020-07-23 10:30:00']. Sau đó cache được đặt lại với thời gian hết hạn là 60 giây mới.
Khi truy cập lần thứ ba trong vòng một phút (ví dụ lúc '2020-07-23 10:30:20'): len(history) = 2, kiểm tra vẫn thành công. Lúc này history=['2020-07-23 10:30:20', '2020-07-23 10:30:10', '2020-07-23 10:30:00']. Sau đó cache được đặt lại với thời gian hết hạn là 60 giây mới.
Khi truy cập lần thứ tư trong vòng một phút (ví dụ lúc '2020-07-23 10:30:30'): len(history) = 3, len(self.history) >= self.num_requests tức là 3 >= 3 đúng, kiểm tra không thành công, trả về False.

Lúc này, chương trình quay lại check_throttles(self, request) ở bước ngoài, gọi phương thức wait() của throttle class, thêm thời gian chờ do phương thức này trả về vào danh sách throttle_durations.

Đi đến phương thức wait(), phương thức này tính toán xem cần chờ bao lâu để có thể truy cập lại.


def wait(self):
    """
    Trả về thời gian đề xuất cho request tiếp theo, tính bằng giây.
    """
    if self.history:
        # Thời gian chờ cho lần truy cập tiếp theo = thời gian hết hạn - (thời gian hiện tại - thời gian history đầu tiên)
        remaining_duration = self.duration - (self.now - self.history[-1])
    else:
        remaining_duration = self.duration

    # Số lần truy cập còn lại = tổng số lần giới hạn - số lần đã truy cập + 1, theo thực tế, kết quả này thường là 1
    available_requests = self.num_requests - len(self.history) + 1
    if available_requests <= 0:
        return None

    # Trả về thời gian chờ cần thiết
    return remaining_duration / float(available_requests)

Quay lại check_throttles(self, request), sau khi có được thời gian wait(), tìm thời gian chờ lớn nhất trong các throttle class duration = max(durations, default=None), sau đó ném ra ngoại lệ, thông báo cần chờ bao nhiêu giây để tiếp tục truy cập.

Trong trường hợp này, nó sẽ trả về cần chờ thêm 30 giây để tiếp tục truy cập.

Khi thời gian đến '2020-07-23 10:31:00', chúng ta truy cập lại. Lúc này history=['2020-07-23 10:30:20', '2020-07-23 10:30:10', '2020-07-23 10:30:00']. Khi đó, điều kiện while self.history and self.history[-1] <= self.now - self.duration sẽ đúng, tức là thời gian history đầu tiên đã qua 60 giây, sẽ thực thi self.history.pop(), loại bỏ thời gian history đầu tiên. Lúc này history=['2020-07-23 10:30:20', '2020-07-23 10:30:10'].

Chương trình tiếp tục, lúc này if len(self.history) >= self.num_requests tức là 2 >= 3 sai, vậy kiểm tra thành công, đi đến throttle_success(), tiếp tục chèn dữ liệu vào history và cache. Lúc này history=['2020-07-23 10:31:00', '2020-07-23 10:30:20', '2020-07-23 10:30:10'].

Nếu lại truy cập ngay lập tức, sẽ lại ném ra ngoại lệ, thời gian chờ là 10 giây.

Kiểm tra lớp permission

Chúng ta định nghĩa cài đặt lớp permission cục bộ trong lớp view: người dùng thông thường tối đa 5 lần truy cập mỗi phút, người dùng ẩn danh tối đa 3 lần truy cập mỗi phút.


class BookAuthViewSet(mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet):
    # Tập hợp kết quả truy vấn
    queryset = Book.objects.filter(is_delete=False)
    # Lớp tuần tự hóa
    serializer_class = BookSerializer
    # Danh sách lớp xác thực
    authentication_classes = [BasicAuthentication, MyAuthentication]
    # Danh sách lớp phân quyền
    permission_classes = [MyPermission, ]
    # Danh sách lớp giới hạn tốc độ/tần suất
    throttle_classes = [UserRateThrottle, AnonRateThrottle]

Tỷ lệ RATE được đặt trong settings.py của dự án:


REST_FRAMEWORK = {
    # Tần suất giới hạn
    'DEFAULT_THROTTLE_RATES': {
        'user': '5/m',
        'anon': '3/m',
    }
}

Kết quả kiểm tra:

Thẻ: django-rest-framework APIView authentication permission throttle

Đăng vào ngày 11 tháng 7 lúc 13:52

Thành phố Cuồng loạn