Skip to content

Instantly share code, notes, and snippets.

@altnight
Created September 19, 2024 06:08
Show Gist options
  • Save altnight/10ec098fdb97380f7975275739dac295 to your computer and use it in GitHub Desktop.
Save altnight/10ec098fdb97380f7975275739dac295 to your computer and use it in GitHub Desktop.
2024/09/19 マルチテナントや権限など、制限されたリソースを取得する際の実装方法

2024/09/19 マルチテナントや権限など、制限されたリソースを取得する際の実装方法

概要/課題/問題意識

  • Webアプリの実装では「ある特定の権限や区分でのみ取得できるリソース」という概念がほぼ確実に存在する
    • 例: 会社Aは会社Bの情報を閲覧できない
    • 例: 情報Aについて、一般ユーザーは閲覧できず、管理者ユーザーは閲覧できる
  • これらの情報の取得処理は、事故ったときのリスクが大きいわりに、実装レベルでは実装者やレビューなどの人力に頼りがち
  • 誰でも簡単に理解できる安全な設計/仕組みが実現できると楽

前提

  • Webアプリ/RDBを使用
  • Django等のActiveRecordベースのORMを含むWebフレームワークを使用

結論

  • Resource(ないしActor)という概念を導入し、取得処理をまとめたPythonのクラスを実装する
  • ビジネスロジックの引数に必ずResource(Actor)を渡す

よくあるリソース制限

  • 会社
    • 例: 会社Aは会社Bの情報を閲覧できない
  • ユーザー権限
    • 例: 所有者/管理者/編集者/一般メンバー
  • 動的な許可
    • 例: アサインされたプロジェクト
    • 例: 許可XXX/担当XXX
    • 例: チーム/グループ
  • アクター
    • 例: システム内部(バッチなどのコマンド)/連携外部システム(外部API)
    • 例: システム管理者(Admin)/運用者(Operator)/顧客(Member)/匿名ユーザー/非ログインユーザー

検討した実装方法

(特になにもしない場合)

全部の記述で書き漏らしがないようにがんばってレビューとかをする方法

  • メリット: レイヤーがないため抽象度が下がり、動く実装を書くだけなら簡単
  • デメリット: 要件が変わったときに人力で抜け漏れをチェックする必要がある
    • セキュリティ対応等で「全部の箇所で問題ないように実装していますか?」と問われたときに、どこまで確信を持って回答できるか
    • 単体テスト自体は書いている想定だが、異常系をあんまりカバーしていないことも多い(レビューの優先度も下げがち)
  • 総評: 結局のところ、実装者/レビュアーがいい感じに全部ちゃんとやってくれるだろうと信頼する手法になる
    • 人間の注意力が試される
    • 小規模なら問題は出にくいと思うが、中規模から大規模で複雑な要件になるほど、どうにかこうにかたまたま動いているという状況になりがち
      • リファクタリングしたいねという話だけ出て進まないイメージ
    • そのうちちょっとしたヘルパー関数的なものを定義して共通処理にしたりするが、それであれば後述のResource(Actor)という概念で記述を統一した方が設計として明確ではある

DjangoのQuerySetやManager

後述のPythonのクラスでの実装パターンと似ているが、フレームワークの機能にのっかる方法

  • メリット: フレームワークの仕組みに近い箇所で実装できる
  • デメリット: どのモデルのQuerySetに処理を紐づけるかを決めるのが微妙。そもそもモデリングとしてモデル(=テーブル)に紐づけるのが適切に見えないことも多い
  • 総評: Webフレームワークが機能として提供していることとアプリの設計で実際に使用するかは別

DBレイヤーでのRowLevelSecurity

DBレイヤーで解決する方法

  • メリット: DBレベルの機能を使用するので、より間違いが起こりにくい
  • デメリット: 適用するまでの手順や手間が若干大掛かり、PostgreSQLが必須
    • 最近の開発だととりあえずPostgreSQLを使用することも多いが、実装者とは別の都合など、もろもろの事情でMySQLになることも全然ある
      • 例: DBの技術選定ができない
      • 例: インフラは別チーム(別会社)が決める
      • 例: リプレイスの都合でMySQLは固定にしておきたい
  • ほか: マルチテナントの設定例で会社ごとに分けたりはするけど、ユーザーの権限など、他の細やかな条件でもカラムを設定するんだろうか?
    • 会社レベルの分離ができていれば大きなリスクは回避できるが、システム的には一般的な他の権限はほぼ存在するので、他は人力でがんばることになるのでは
  • 総評: 導入の敷居が高く実装レベルで完結しない。また、仕組みをやめる場合にも若干大掛かりになる

Repositoryのコンストラクタでbindする

いわゆるクリーンアーキテクチャ(諸説あり)のリポジトリのコンストラクタに渡して必要な値を束縛する方法

ただ、(DBアクセスの)Repositoryパターンの実装といっても、各人の想像する実装方法がバラバラでふんわりとした話にはなる。一応今回はテーブルと1:1で対応したRepositoryがあるパターンを想定。

  • メリット: 考え方としては筋が通っている
  • デメリット: ActiveRecordベースのORMを採用しているWebフレームワークの場合、実装としては冗長になり、結局トータルでシステム全体で見ると保守性が下がる可能性が高い(個人的見解)
  • 総評: デメリットが大きいため、別の手法にしたい

素のPythonのクラス

上記を踏まえた、現状の開発でやっている手法

  • メリット:
    • 純粋なPythonのクラスで依存がないため、比較的簡素な仕組み
      • 仕組みをやめるのも比較的簡単
    • 型チェックも同時に実施することで、「この処理は必ずこのResource(Actor)が実行する」ことが保証できる
      • コードの表現力が上がり、可読性が上がる
      • 逆に言うと「(主語となる)誰が呼ぶ想定のビジネスロジックなのか」がわからない実装を排除できる
  • デメリット:
    • 抽象レイヤーが1つ増える
      • 設計はシステム全体で守られないと意味がなくなるため、レビューで確実に指摘する必要がある
        • オレオレフレームワークではないが、オレオレ設計ではある
    • システム内部(コマンドなど)を表すResource(Actor)の処理で毎回無制限リソースを渡すことになり、若干記述が冗長な箇所もある
  • 総評: うまくいかなかった場合にやめることも込み込みで、現実解としては十分だと思う

備考

  • 強制力を持たせるにはabcなどの抽象レイヤーをいれる必要があるが、現状はそこまでやっていない
    • 強制力の強い実装をいれるほどの確信はまだない
    • 実際の実装にいい感じに全部対応できるようないい感じ実装しないと、逆に面倒になりそう
  • Django的にはQuerySetを返すようにしているため、遅延評価されるため、クエリ数の増加はそこまで気にしなくてよい
    • レイヤーを増やしたことで1,2個クエリが増えるかもしれないが、安全側に倒せるメリットを優先するでよい
  • 個別の引数を渡したり、select_related等の指定はしない
    • 遅延評価なので、N+1が発生する箇所に近い箇所で記述した方が適切
    • ビジネスロジックとしての絞り込みはビジネスロジックを扱うモジュールで記述する方が適切
  • よくある注意事項: 引数に具体的な絞り込んだ後のオブジェクトを渡さない
    • ビジネスロジックから見て、外側の値を信頼しないことが重要
    • 必ずResource(Actor)が渡されて、そのResource(Actor)が制限された必要な情報を取得するから安全なのであって、直接値を渡したら再度検証が必要になり、仕組みとしての意味がなくなる

サンプルコード

  • 顧客は会社に所属している
  • プロジェクトは会社が作成する
  • プロジェクトのアサインは会社と顧客の中間テーブル(N:N)
# models.py
# 記述を短くするため、本来動作させるために必要な引数を省略している
class AuthUser(models.Model):
    is_staff = models.BooleanField()
    is_superuser = models.BooleanField()

class Company(models.Model):
    disabled = models.BooleanField()

class Customer(models.Model):
    auth_user = models.OneToOneField(AuthUser)
    company = models.ForeginKey(Company)
    
class Project(models.Model):
    name = models.CharField()
    company = models.ForeginKey(Company)

class ProjectAssign(models.Model):
    project = models.ForeginKey(Project)
    customer = models.ForeginKey(Customer)
    
# resources.py

# SystemCommandResourceとかの命名の方がいいかもしれない
# CommandActorとか
class UnlimitedResource:
    """無制限リソース"""

class CompanyResource:
    """会社リソース"""
    def __init__(self, company: Company):
        if company.disabled:
            raise
        self.company = company
        
    def get_projects(self) -> QuerySet[Project]:
        return self.company.project_set.all()
        # return Project.objects.filter(company=self.company)
    
    def get_project(self, project_id: int) -> Project:
        return self.get_projects().get(id=project_id)
    
    def get_customers(self) -> QuerySet[Customer]:
        return self.company.customer_set.all()
        # return Customer.objects.filter(company=self.company)
    
    def get_customer(self, customer_id: int) -> Customer:
        return self.get_customers().get(id=customer_id)
        

class CustomerResource:
    """顧客(一般ユーザー)リソース"""
    def __init__(self, customer: Customer):
        if customer.auth_user.is_staff or customer.auth_user.is_superuser:
            raise
        self.customer = customer
        
    @property
    def company(self) -> Company:
        return self.customer.company
    
    @property
    def company_resource(self) -> CompanyResource:
        return CompanyResource(self.company)
        
    def get_projects1(self) -> QuerySet[Project]:
        # CompanyResourceを通す場合。別の箇所で整合性をとれてるなら(2)でもよい
        company_projects = self.company_resource.get_projects()
        assigned_project_ids = ProjectAssign.objects.filter(
            project__in=company_projects,
            customer=self.customer,
        ).values_list("project", flat=True)
        return Project.objects.filter(id__in=assigned_project_ids)
    
    def get_projects2(self) -> QuerySet[Project]:
        # CompanyResourceを通さない場合
        assigned_project_ids = ProjectAssign.objects.filter(
            customer=self.customer,
        ).values_list("project", flat=True)
        return Project.objects.filter(id__in=assigned_project_ids)
    
    def get_project(self, project_id: int) -> Project:
        return self.get_projects1().get(id=project_id)

        
class AdminCustomerResource(CustomerResource):
    """顧客(管理者)リソース"""
    def __init__(self, customer: Customer):
        if not customer.auth_user.is_staff:
            raise
        self.customer = customer

    def get_projects(self) -> QuerySet[Project]:
        return self.company_resource.get_projects()
    
    def get_project(self, project_id: int) -> QuerySet[Project]:
        return self.company_resource.get_project(project_id=project_id)
# api.py
@api.post("/api/project/{project_id}/update")
def api_update_project_name(request, project_id: int, payload: dict):
    services.update_project_name(
        resource=CustomerResource(request.user),
        project_id=project_id,
        value=payload["value"],
    )

# services.py
# 関数での実装例
def update_project_name(resource: "CustomerResource", project_id: int, value: str):
    project = resource.get_project(project_id)
    project.name = value
    project.save()

# クラスでの実装例
class CustomerProjectOperation:
    def __init__(self, resource: "CustomerResource", project_id: int):
        self.resource = resource
        self.project = resource.get_project(project_id=project_id)

    def update_name(self, value: str) -> None:
        self.project.name = value
        self.project.save()

    def list(self) -> QuerySet[Project]:
        return self.resource.get_projects().order_by("-id")

# NG例: 外側の値を信頼してしまっている
# api.py
@api.post("/api/project/{project_id}/update")
def api_update_project_name(request, project_id: int, payload: dict):
    resource = CustomerResource(request.user)
    project = resource.get_project(project_id)
    services.update_project_attr(
        project=project,
        value=payload["value"],
    )

# services.py
def update_project_name(project: Project, value: str):
    # ここにやってきたProjectが信頼できない。再チェックをするなら意味がない
    project.name = value
    project.save()
# commands.py
class Command(BaseCommand):
    def handle(self):
        disable_company(
            resource=UnlimitedResource(),
        )
        
# services.py
def update_xxx(resource: "UnlimitedResource"):
    # 引数にリソースを渡すことで、型チェックを通したい。実際のクラス自体は空なのでなにもしない
    # 無制限リソースに関してはこういった記述が増えてしまうので、若干冗長。なんらかの意味づけを増やすなりしたいところ 
    assert resource
    Company.objects.all().update(disabled=True)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment