如何在Flask中实现基于角色的访问控制?

10
有没有维护良好的插件可以帮助我创建一个具有基于角色的访问控制(Flask app with role based access control)的Flask应用程序?例如,管理角色(admin role),会计角色(accounting role),人力资源角色(hr role)等等。
看起来Flask-User不错,但是这些讨论表明维护者已经离开了... https://gitter.im/Flask-User/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge Flask-Login需要Flask-Security,但是后者未得到维护,不过可以使用Flask-Security-Too。后者通过Flask-Principal实现授权,但是其最后一次发布是在2013年,对我来说似乎很不活跃(https://github.com/mattupstate/flask-principal/issues/50)。
感谢任何建议。

这可能是一个选项。https://pypi.org/project/Flask-Authorize/ - Kevin Crick
1
谢谢 - 这肯定是一个可行的选择。 我曾经被一些只有一人维护并最终放弃的插件烧过几次,但没有风险就没有回报。 我会再等一段时间,也许会有更多的答案。 该项目仅于十月初开始。 - Jürgen Gmach
4个回答

4

也有可能没有任何一种解决方案是100%合适的,即您需要进行更改或扩展。

在这种情况下,我的建议是从基本模型开始,根据您的需求进行扩展。 某些插件的想法和方法可以帮助您解决问题。 这种方法的优点是,您将学到更多关于底层发生了什么,而不是只从某些插件中导入。

插件很棒,社区已经付出了很多努力扩展功能。 没有任何疑问。

把答案看作是动力,因为您的问题可以从最小代码版本0.0.1开始:

class RolesUsers(Base):
    __tablename__ = 'roles_users'

    id = db.Column(db.Integer(), primary_key=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id'))
    role_id = db.Column(db.Integer(), db.ForeignKey('role.id'))

class Role(RoleMixin, Base):
    __tablename__ = 'role'

    id = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.String(80), unique=True)

    def __repr__(self):
        return self.name

class User(UserMixin, Base):
    __tablename__ = 'user'

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), index=True, unique=True)

    roles = db.relationship('Role', secondary='roles_users',
                            backref=db.backref('users', lazy='dynamic'))

请查看M2M

更新:来自评论的请求

现在你有一个小型应用程序,代表了提到的第一个版本0.0.1 :) 由于你有一段时间没有使用FLASK了,我尽力通过注释提醒你一些基本细节。

import datetime
from functools import wraps
from flask import Flask, redirect, url_for, session, render_template_string
from flask_sqlalchemy import SQLAlchemy


# Class-based application configuration
class ConfigClass(object):
    """ Flask application config """

    # Flask settings
    SECRET_KEY = 'This is an INSECURE secret!! DO NOT use this in production!!'

    # Flask-SQLAlchemy settings
    SQLALCHEMY_DATABASE_URI = 'sqlite:///db.sqlite'    # File-based SQL database
    SQLALCHEMY_TRACK_MODIFICATIONS = False    # Avoids SQLAlchemy warning

def create_app():
    """ Flask application factory """
    
    # Create Flask app load app.config
    app = Flask(__name__)
    app.config.from_object(__name__+'.ConfigClass')

    # Initialize Flask-SQLAlchemy
    db = SQLAlchemy(app)

    @app.before_request
    def before_request():
        try:
            print("Current Role: ", session['role'])
        except:
            print("Current Role: Guest")

    # Define the User data-model.
    class User(db.Model):
        __tablename__ = 'users'
        id = db.Column(db.Integer, primary_key=True)
        active = db.Column('is_active', db.Boolean(), nullable=False, server_default='1')

        # User authentication information. The collation='NOCASE' is required
        # to search case insensitively when USER_IFIND_MODE is 'nocase_collation'.
        email = db.Column(db.String(255, collation='NOCASE'), nullable=False, unique=True)
        email_confirmed_at = db.Column(db.DateTime())
        password = db.Column(db.String(255), nullable=False, server_default='')

        # User information
        first_name = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='')
        last_name = db.Column(db.String(100, collation='NOCASE'), nullable=False, server_default='')

        # Define the relationship to Role via UserRoles
        roles = db.relationship('Role', secondary='user_roles')

    # Define the Role data-model
    class Role(db.Model):
        __tablename__ = 'roles'
        id = db.Column(db.Integer(), primary_key=True)
        name = db.Column(db.String(50), unique=True)

    # Define the UserRoles association table
    class UserRoles(db.Model):
        __tablename__ = 'user_roles'
        id = db.Column(db.Integer(), primary_key=True)
        user_id = db.Column(db.Integer(), db.ForeignKey('users.id', ondelete='CASCADE'))
        role_id = db.Column(db.Integer(), db.ForeignKey('roles.id', ondelete='CASCADE'))

    # First Drop then Create all database tables
    db.drop_all()
    db.create_all()

    # Create 'member@example.com' user with no roles
    if not User.query.filter(User.email == 'member@example.com').first():
        user = User(
            email='member@example.com',
            email_confirmed_at=datetime.datetime.utcnow(),
            password='Password1',
        )
        db.session.add(user)
        db.session.commit()

    # Create 'admin@example.com' user with 'Admin' and 'Agent' roles
    if not User.query.filter(User.email == 'admin@example.com').first():
        user = User(
            email='admin@example.com',
            email_confirmed_at=datetime.datetime.utcnow(),
            password='Password1',
        )
        user.roles.append(Role(name='Admin'))
        user.roles.append(Role(name='Member'))
        db.session.add(user)
        db.session.commit()

    def access_required(role="ANY"):
        """
        see: https://flask.palletsprojects.com/en/2.1.x/patterns/viewdecorators/
        """
        def wrapper(fn):
            @wraps(fn)
            def decorated_view(*args, **kwargs):
                if session.get("role") == None or role == "ANY":
                    session['header'] = "Welcome Guest, Request a new role for higher rights!"
                    return redirect(url_for('index'))
                if session.get("role") == 'Member' and role == 'Member':
                    print("access: Member")
                    session['header'] = "Welcome to Member Page!"
                    return redirect(url_for('index'))
                if session.get("role") == 'Admin' and role == 'Admin':
                    session['header'] = "Welcome to Admin Page!"
                    print("access: Admin")
                else:
                    session['header'] = "Oh no no, you haven'tn right of access!!!"
                    return redirect(url_for('index'))
                return fn(*args, **kwargs)
            return decorated_view
        return wrapper

    # The index page is accessible to anyone
    @app.route('/')
    def index():
        print("index:", session.get("role", "nema"))
        return render_template_string('''
                <h3>{{ session["header"] if session["header"] else "Welcome Guest!" }}</h3>
                <a href="/admin_role">Get Admin Role</a> &nbsp&nbsp | &nbsp&nbsp
                <a href="/admin_page">Admin Page</a> <br><br>
                <a href="/member_role">Get Member Role</a> &nbsp&nbsp | &nbsp&nbsp
                <a href="/member_page">Member Page</a> <br><br>
                <a href="/guest_role">Get Guest Role</a> <br><br>
                <a href="/data">Try looking at DATA with different roles</a>
            ''')

    @app.route('/data')
    def data():
        return render_template_string("""
                <a href="/admin_role">Admin Role</a>
                <a href="/member_role">Member Role</a>
                <a href="/guest_role">Guest Role</a>
                <br><p>The data page will display a different context depending on the access rights.</p><br> 
                {% if not session['role'] %}
                    <h2>You must have diffrent role for access to the data.</h2>
                    <a href="{{ url_for('.index') }}">Go back?</a>
                {% endif %}

                {% if session['role'] == 'Member' or  session['role'] == 'Admin' %}
                <h2>USERS:</h2>
                <table>
                <tr>
                    <th>ID</th>
                    <th>EMAIL</th>
                </tr>
                {% for u in users %}
                <tr>
                    <td>{{ u.id }}</td>
                    <td>{{ u.email }}</td>
                </tr>
                {% endfor %}
                </table>
                {% endif %}

                {% if session['role'] == 'Admin' %}
                    <h2>ROLE:</h2>
                    <table>
                    <tr>
                        <th>ID</th>
                        <th>NAME</th>
                    </tr>
                    {% for r in roles %}
                    <tr>
                        <td>{{ r.id }}</td>
                        <td>{{ r.name }}</td>
                    </tr>
                    {% endfor %}
                    </table>

                    <h2>USER ROLES:</h2>
                    <table>
                    <tr>
                        <th>ID</th>
                        <th>USER ID</th>
                        <th>ROLE ID</th>
                    </tr>
                    {% for ur in user_roles %}
                    <tr>
                        <td>{{ ur.id }}</td>
                        <td>{{ ur.user_id }}</td>
                        <td>{{ ur.role_id }}</td>
                    </tr>
                    {% endfor %}
                    </table>
                {% endif %}
            """, 
                users=User.query, 
                roles= Role.query, 
                user_roles=UserRoles.query)

    @app.route("/member_role")
    def member_role():
        """
        Anyone can access the url and get the role of a MEMBER.
        """
        r = Role.query.get(2)
        session['role'] = r.name
        session['header'] = "Welcome to Member Access!"
        return render_template_string('<h2>{{ session["header"] }}</h2> <a href="/">Go back?</a>')

    @app.route("/member_page")
    @access_required(role="Member")
    def member_page():
        # session['header'] = "Welcome to Admin Page!"
        return render_template_string('<h1>{{ session["header"] }}</h1> <a href="/">Go back?</a>')

    @app.route("/admin_role")
    def admin_role():
        """
        Anyone can access the url and get the role of a ADMIN.
        """
        r = Role.query.get(1)
        session['role'] = r.name
        session['header'] = "Welcome to Admin Access!"
        return render_template_string('<h1>{{ session["header"] }}</h1> <a href="/">Go back?</a>')

    @app.route("/admin_page")
    @access_required(role="Admin")
    def admin_page():
        """
        This url requires an ADMIN role.
        """
        return render_template_string('<h1>{{ session["header"] }}</h1> <a href="/">Go back?</a>')

    @app.route("/guest_role")
    def guest_role():
        """
        For the GUEST, we will only delete the session and thus 'kill' the roles.
        """
        session.clear()
        return redirect(url_for('index'))

    return app


# Start development web server
if __name__ == '__main__':
    app = create_app()
    app.run(host='0.0.0.0', port=5000, debug=True)


现在我们可以开始游戏了,因为有一些规则,包括:只有具有管理员角色的用户才能访问管理员页面,成员和他的页面也是如此。此外,您有一个由第一个答案版本中提到的模型组成的数据列表。这个数据列表也可以用不同的角色查看。细节:如果您拥有管理员角色,则可以看到所有要显示的数据;如果您拥有成员角色,则只能查看用户,并且如果您是访客,则无法查看任何数据。

下面是它的图示形式。

enter image description here enter image description here enter image description here enter image description here enter image description here enter image description here

我希望这个简单的例子能够对你和其他stackcoverflow社区成员的进一步学习有所帮助。我坚信你很快就会拥有一个更好的版本。
愉快的编程...

谢谢。你能加上导入吗?我已经有一段时间没有使用Flask了,虽然它们看起来很熟悉,但我不记得它们来自哪里了。谢谢! - Jürgen Gmach
好的,现在你有了一个非常简单但足以理解我在第一个答案版本中所起始的本质的例子。 - Milovan Tomašević

4

我正在同样的旅程中。 你可以看一下flask-RBAC => https://flask-rbac.readthedocs.io/ 其中RBAC代表"基于角色的访问控制"...

最后一次提交是不到30天之前。

我不够专业,无法确认它是否比你提到的那些更好,但考虑到其如此明确的名称,可能值得一试。 让我知道你对此有何想法。

(PS: 我与这个项目没有任何关联)


谢谢你的提示!目前我没有新的Flask项目,所以我无法尝试flask-rbac,但我会记住的。再次感谢! - Jürgen Gmach

2
对于在2022年偶然发现这篇文章的人,我想提供一种替代方案,以便帮助任何人!与其自己开发授权系统,你可以考虑将其与应用完全解耦,从而避免构建自己系统时遇到的大量复杂性。 Cerbos是一个开源项目,它允许你定义和部署访问策略到单独的服务,并暴露一个简单的API来检查权限是否与策略决策点(PDP)一致。PDP是单独部署的(目前通过kube svc / sidecar、systemd服务或AWS lambdas)。 它可以通过SDK轻松地与Python集成。创建策略看起来像这样:例如角色“管理员”可以对“联系人”进行任何操作,角色“用户”只能读取“联系人”。
---
apiVersion: api.cerbos.dev/v1
resourcePolicy:
  version: default
  resource: contact
  rules:
    - actions: ["*"]
      effect: EFFECT_ALLOW
      roles:
        - admin

    - actions: ["read"]
      effect: EFFECT_ALLOW
      derivedRoles:
        - user

然后在代码中检查访问权限:
user = Principal(
    "some_user_id",
    roles={"user"},  # retrieved from IdP/data store
)

contact = Resource(
    id="some_contact_id",
    kind="contact",
)

with CerbosClient(host="http://localhost:3592") as c:
    is_allowed = c.is_allowed("read", user, contact)

一个更加深入的例子(使用 FastAPI 而非 Flask 和 SQLAlchemy)在这里,如果您感兴趣:在 SQLAlchemy 中使用 Cerbos 实现基于角色和属性的访问控制 (完全透明披露:我在 Cerbos 工作!)

0

嘿,如果你还没有开始你的项目,可以查看Flask-Authorize


网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接