Django內置權限擴展案例

當Django的內置權限沒法知足需求的時候就本身擴展吧~

背景介紹

overmind項目使用了Django內置的權限系統，Django內置權限系統基於model層作控制，新的model建立後會默認新建三個權限，分別爲：add、change、delete，若是給用戶或組賦予delete的權限，那麼用戶將能夠刪除這個model下的全部數據。

本來overmind只管理了咱們本身部門的數據庫，權限設置只針對具體的功能不針對細粒度的數據庫實例，例如用戶A 有審覈的權限，那麼用戶A 能夠審覈全部的DB，此時使用內置的權限系統就能夠知足需求了，但隨着系統的不斷完善要接入其餘部門的數據庫管理，這就要求針對不一樣用戶開放不一樣DB的權限了，例如A部門的用戶只能操做A部門的DB，Django內置基於model的權限沒法知足需求了。

實現過程

先來肯定下需求：

1. 保持本來的基於功能的權限控制不變，例如用戶A有查詢權限，B有審覈權限
2. 增長針對DB實例的權限控制，例如用戶A只能查詢特定的DB，B只能審覈特定的DB

對於上邊需求1用內置的權限系統已經能夠實現，這裏不贅述，重點看下需求2，DB信息都存放在同一個表裏，不一樣用戶能操做不一樣的DB，也就是須要把每一條DB信息與有權限操做的用戶進行關聯，爲了方便操做，咱們考慮把DB跟用戶組關聯，在用戶組裏的用戶都有權限，而操做類型通過分析主要有兩類讀和寫，那麼須要給每一個MySQL實例添加兩個字段分別記錄對此實例有讀和寫權限的用戶組

以下代碼在原來的model基礎上添加read_groups和write_groups字段，DB實例跟用戶組應是ManyToManyField多對多關係，一個實例能夠關聯多個用戶組，一個用戶組也能夠屬於多個實例

class Mysql(models.Model):
    Env = (
        (1, 'Dev'),
        (2, 'Qa'),
        (3, 'Prod'),
    )
    create_time = models.DateTimeField(auto_now_add=True, verbose_name='建立時間')
    update_time = models.DateTimeField(auto_now=True, verbose_name='更新時間')

    project_id = models.IntegerField(verbose_name='項目')
    project_tmp = models.CharField(max_length=128, default='')
    environment = models.IntegerField(choices=Env, verbose_name='環境')

    master_host = models.GenericIPAddressField(verbose_name='master主機')
    master_port = models.IntegerField(default=3306, verbose_name='master端口')

    slave_host = models.GenericIPAddressField(null=True, verbose_name='slave主機')
    slave_port = models.IntegerField(null=True, default=3306, verbose_name='slave端口')

    database = models.CharField(max_length=64, verbose_name='數據庫')

    read_groups = models.ManyToManyField(Group, related_name='read', verbose_name='讀權限')
    write_groups = models.ManyToManyField(Group, related_name='write', verbose_name='寫權限')

    description = models.TextField(null=True, verbose_name='備註')
複製代碼

model肯定了，接下來咱們分三部分詳細介紹下權限驗證的具體實現

列表頁權限控制

如上圖列表頁，每一個用戶進入系統後只能查看本身有讀權限的MySQL實例列表，管理員能查看全部，代碼以下：

def mysql(request):
    if request.method == 'GET':
        if request.user.is_superuser:
            _lists = Mysql.objects.all().order_by('id')
        else:
            # 獲取登陸用戶的全部組
            _user_groups = request.user.groups.all()

            # 構造一個空的QuerySet而後合併
            _lists = Mysql.objects.none()
            for group in _user_groups:
                _lists = _lists | group.read.all()

        return render(request, 'overmind/mysql.index.html', {'request': request, 'lPage': _lists})
複製代碼

實現的思路是：獲取登陸用戶的全部組，而後循環查詢每一個組有讀取權限的數據庫實例，最後把每一個組有權限讀的數據庫實例進行合併返回

獲取登陸用戶的全部組用到了ManyToMany的查詢方法：request.user.groups.all()

最終返回的一個結果是QuerySet，因此咱們須要先構造一個空的Queryset：Mysql.objects.none()

QuerySet合併不能用簡單的相加，應爲：QuerySet-1 | QuerySet-2

查詢接口權限控制

如上圖系統中有不少功能是須要根據項目、環境查詢對應的DB信息的，對於此類接口也須要控制用戶只能查詢本身有權限讀的DB實例，管理員能查看全部，代碼以下：

def get_project_database(request, project, environment):
    if request.method == 'GET':
        _jsondata = {}

        if request.user.is_superuser:
            # 返回全部項目和環境匹配的DB
            _lists = Mysql.objects.filter(
                project_id=int(project),
                environment=int(environment)
            )

            _jsondata = {i.id: i.database for i in _lists}
        else:
            # 只返回用戶有權限查詢的DB
            _user_groups = request.user.groups.all()

            for group in _user_groups:
                # 循環mysql表中有read_groups權限的全部組
                for mysql in group.read.all():
                    if mysql.project_id == int(project) and mysql.environment == int(environment):
                        _jsondata[mysql.id] = mysql.database

        return JsonResponse(_jsondata)
複製代碼

實現思路與上邊相似，只是多了一步根據項目和環境再進行判斷

須要根據group去反查都有哪些DB實例包含了該組，這裏用到了M2M的related_name屬性：group.read.all()

更多關於Django ORM查詢的內容能夠看這篇文章Django model select的各類用法詳解有詳細的總結

執行操做權限控制

除了上邊的兩個場景以外咱們還須要在執行具體的操做以前去判斷是否有權限，例如執行審覈操做前判斷用戶是否對此DB有寫權限

有不少地方都須要作這個判斷，因此把這個權限判斷單獨寫個方法來處理，代碼以下：

def check_permission(perm, mysql, user):
    # 若是用戶是超級管理員則有權限
    if user.is_superuser:
        return True

    # 取出用戶所屬的全部組
    _user_groups = user.groups.all()

    # 取出Mysql對應權限的全部組
    if perm == 'read':
        _mysql_groups = mysql.read_groups.all()
    if perm == 'write':
        _mysql_groups = mysql.write_groups.all()

    # 用戶組和DB權限組取交集，有則表示有權限，不然沒有權限
    group_list = list(set(_user_groups).intersection(set(_mysql_groups)))

    return False if len(group_list) == 0 else True
複製代碼

實現思路是：根據傳入的第三個用戶參數，來獲取到用戶全部的組，而後根據傳入的第一個參數類型讀取或寫入和第二個參數DB實例來獲取到有權限的全部組，而後對兩個組取交集，交集不爲空則表示有權限，爲空則沒有

M2M的.all()取出來的結果是個list，兩個list取交集的方法爲：list(set(list-A).intersection(set(list-B)))

view中使用就很簡單了，以下：

def query(request):
    if request.method == 'POST':
        postdata = request.body.decode('utf-8')
        _host = get_object_or_404(Mysql, id=int(postdata.get('database')))

        # 檢查用戶是否有DB的查詢權限
        if check_permission('read', _host, request.user) == False:
            return JsonResponse({'state': 0, 'message': '當前用戶沒有查詢此DB的權限'})
複製代碼

寫在最後

1. Django有第三方的基於object的權限管理模塊Django-guardian，本項目沒有使用主要是由於一來權限需求並不複雜，本身實現也很方便，二來我的在非必要的狀況下並不喜歡引用過多第三方的包，後續升級維護都是負擔
2. 方案和代碼不盡完美，各位有更好的方案建議或更優雅的代碼寫法歡迎與我交流

---

若是你以爲文章不錯，請點右下角【好看】。若是你以爲讀的不盡興，推薦閱讀如下文章：

• Django+JWT實現Token認證
• 咱們自研的那些Devops工具