prowler-api

使用する場面

Prowler 固有のパターンには、このスキルを使用します。

• 行レベルセキュリティ（RLS）／テナント分離
• RBAC 権限とロールチェック
• プロバイダーのライフサイクルと検証
• テナントコンテキストを使用する Celery タスク
• マルチデータベースアーキテクチャ（4 データベース構成）

一般的な DRF パターン（ViewSets、Serializers、Filters、JSON:API）には、django-drf スキルを使用します。

---

重要なルール

• ViewSet のコンテキスト外でクエリを実行する場合は、必ず rls_transaction(tenant_id) を使用します
• 権限を確認する前に、必ず get_role() を使用します（最初のロールのみを返します）
• デコレーターの順序は必ず @set_tenant、次に @handle_provider_deletion とします
• M2M リレーションシップでは、必ず明示的な through モデルを使用します（RLS に必要です）
• Celery タスクで、RLS コンテキストなしに Provider.objects にアクセスしてはいけません
• 生 SQL や connection.cursor() を使用して、RLS を回避してはいけません
• Django のデフォルト M2M を使用してはいけません。RLS では tenant_id を持つ through モデルが必要です

注：rls_transaction() は UUID オブジェクトと文字列の両方を受け付けます。内部で str(value) を使って変換します。

---

アーキテクチャの概要

4 データベースアーキテクチャ

データベース	エイリアス	用途	RLS
default	prowler_user	標準 API クエリ	はい
admin	admin	マイグレーション、認証のバイパス	いいえ
replica	prowler_user	読み取り専用クエリ	はい
admin_replica	admin	管理者用読み取りレプリカ	いいえ

# When to use admin (bypasses RLS)
from api.db_router import MainRouter
User.objects.using(MainRouter.admin_db).get(id=user_id)  # Auth lookups

# Standard queries use default (RLS enforced)
Provider.objects.filter(connected=True)  # Requires rls_transaction context

RLS トランザクションのフロー

Request → Authentication → BaseRLSViewSet.initial()
                                    │
                                    ├─ Extract tenant_id from JWT
                                    ├─ SET api.tenant_id = 'uuid' (PostgreSQL)
                                    └─ All queries now tenant-scoped

---

実装チェックリスト

Prowler 固有の API 機能を実装する場合：

#	パターン	参照先	要点
1	RLS モデル	api/rls.py	RowLevelSecurityProtectedModel を継承し、制約を追加します
2	RLS トランザクション	api/db_utils.py	rls_transaction(tenant_id) コンテキストマネージャーを使用します
3	RBAC 権限	api/rbac/permissions.py	get_role()、get_providers()、Permissions 列挙型
4	プロバイダーの検証	api/models.py	Provider モデルの validate_<provider>_uid() メソッド
5	Celery タスク	tasks/tasks.py、api/decorators.py、config/celery.py	タスク定義、デコレーター（@set_tenant、@handle_provider_deletion）、RLSTask 基底クラス
6	RLS シリアライザー	api/v1/serializers.py	RLSSerializer を継承して tenant_id を自動挿入します
7	Through モデル	api/models.py	すべての M2M で、tenant_id を持つ明示的な through モデルを使用する必要があります

ファイルの完全なパス：references/file-locations.md を参照してください

---

判断フロー

どの基底モデルを使うか？

Tenant-scoped data       → RowLevelSecurityProtectedModel
Global/shared data       → models.Model + BaseSecurityConstraint (rare)
Partitioned time-series  → PostgresPartitionedModel + RowLevelSecurityProtectedModel
Soft-deletable           → Add is_deleted + ActiveProviderManager

どのマネージャーを使うか？

Normal queries           → Model.objects (excludes deleted)
Include deleted records  → Model.all_objects
Celery task context      → Must use rls_transaction() first

どのデータベースを使うか？

Standard API queries     → default (automatic via ViewSet)
Read-only operations     → replica (automatic for GET in BaseRLSViewSet)
Auth/admin operations    → MainRouter.admin_db
Cross-tenant lookups     → MainRouter.admin_db (use sparingly!)

Celery タスクのデコレーターの順序は？

@shared_task(base=RLSTask, name="...", queue="...")
@set_tenant                    # First: sets tenant context
@handle_provider_deletion      # Second: handles deleted providers
def my_task(tenant_id, provider_id):
    pass

---

RLS モデルのパターン

from api.rls import RowLevelSecurityProtectedModel, RowLevelSecurityConstraint

class MyModel(RowLevelSecurityProtectedModel):
    # tenant FK inherited from parent
    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
    name = models.CharField(max_length=255)
    inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
    updated_at = models.DateTimeField(auto_now=True, editable=False)

    class Meta(RowLevelSecurityProtectedModel.Meta):
        db_table = "my_models"
        constraints = [
            RowLevelSecurityConstraint(
                field="tenant_id",
                name="rls_on_%(class)s",
                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
            ),
        ]

    class JSONAPIMeta:
        resource_name = "my-models"

M2M リレーションシップ（必ず through モデルを使用）

class Resource(RowLevelSecurityProtectedModel):
    tags = models.ManyToManyField(
        ResourceTag,
        through="ResourceTagMapping",  # REQUIRED for RLS
    )

class ResourceTagMapping(RowLevelSecurityProtectedModel):
    # Through model MUST have tenant_id for RLS
    resource = models.ForeignKey(Resource, on_delete=models.CASCADE)
    tag = models.ForeignKey(ResourceTag, on_delete=models.CASCADE)

    class Meta:
        constraints = [
            RowLevelSecurityConstraint(
                field="tenant_id",
                name="rls_on_%(class)s",
                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
            ),
        ]

---

非同期タスクの応答パターン（202 Accepted）

実行時間の長い操作では、タスク参照とともに 202 を返します。

@action(detail=True, methods=["post"], url_name="connection")
def connection(self, request, pk=None):
    with transaction.atomic():
        task = check_provider_connection_task.delay(
            provider_id=pk, tenant_id=self.request.tenant_id
        )
    prowler_task = Task.objects.get(id=task.id)
    serializer = TaskSerializer(prowler_task)
    return Response(
        data=serializer.data,
        status=status.HTTP_202_ACCEPTED,
        headers={"Content-Location": reverse("task-detail", kwargs={"pk": prowler_task.id})}
    )

---

プロバイダー（11 種類をサポート）

プロバイダー	UID の形式	例
AWS	12 桁	123456789012
Azure	UUID v4	a1b2c3d4-e5f6-...
GCP	6～30 文字、小文字、先頭は英字	my-gcp-project
M365	有効なドメイン	contoso.onmicrosoft.com
Kubernetes	2～251 文字	arn:aws:eks:...
GitHub	1～39 文字	my-org
IaC	Git URL	https://github.com/user/repo.git
Oracle Cloud	OCID 形式	ocid1.tenancy.oc1..
MongoDB Atlas	24 文字の 16 進数	507f1f77bcf86cd799439011
Alibaba Cloud	16 桁	1234567890123456

新しいプロバイダーの追加：ProviderChoices 列挙型に追加し、validate_<provider>_uid() 静的メソッドを作成します。

---

RBAC 権限

権限	制御対象
MANAGE_USERS	ユーザーの CRUD、ロール割り当て
MANAGE_ACCOUNT	テナント設定
MANAGE_BILLING	請求／サブスクリプション
MANAGE_PROVIDERS	プロバイダーの CRUD
MANAGE_INTEGRATIONS	インテグレーション設定
MANAGE_SCANS	スキャンの実行
UNLIMITED_VISIBILITY	すべてのプロバイダーを表示（provider_groups をバイパス）

RBAC の可視性パターン

def get_queryset(self):
    user_role = get_role(self.request.user)
    if user_role.unlimited_visibility:
        return Model.objects.filter(tenant_id=self.request.tenant_id)
    else:
        # Filter by provider_groups assigned to role
        return Model.objects.filter(provider__in=get_providers(user_role))

---

Celery キュー

キュー	用途
scans	Prowler のスキャン実行
overview	ダッシュボードの集計（重大度、攻撃対象領域）
compliance	コンプライアンスレポートの生成
integrations	外部連携（Jira、S3、Security Hub）
deletion	プロバイダー／テナントの削除（非同期）
backfill	過去データのバックフィル処理
scan-reports	出力の生成（CSV、JSON、HTML、PDF）

---

タスクの構成（Canvas）

複雑なワークフローには、Celery の Canvas プリミティブを使用します。

プリミティブ	用途
chain()	順次実行：A → B → C
group()	並列実行：A、B、C を同時に実行
組み合わせ	複雑なワークフロー向けに、ネストしたグループを含むチェーン

注： 結果の受け渡しを防ぐには .si()（シグネチャ不変）を使用します。結果を受け渡す必要がある場合は .s() を使用します。

例： チェーン、グループ、組み合わせのパターンについては、assets/celery_patterns.py を参照してください。

---

Beat スケジューリング（定期タスク）

操作	要点
スケジュールを作成	IntervalSchedule.objects.get_or_create(every=24, period=HOURS)
定期タスクを作成	関数ではなくタスク名を使用し、kwargs=json.dumps(...) を指定します
スケジュール済みタスクを削除	PeriodicTask.objects.filter(name=...).delete()
競合状態を回避	DB のコミットを待つために countdown=5 を使用します

例： schedule_provider_scan パターンについては、assets/celery_patterns.py を参照してください。

---

高度なタスクパターン

@set_tenant の動作

モード	kwargs 内の tenant_id	関数に渡される tenant_id
@set_tenant（デフォルト）	ポップ（削除）されます	いいえ — 関数には渡されません
@set_tenant(keep_tenant=True)	読み取られますが保持されます	はい — 関数に渡されます

主なパターン

パターン	説明
bind=True	self.request.id、self.request.retries にアクセスします
get_task_logger(__name__)	Celery タスクで適切なログを記録します
SoftTimeLimitExceeded	強制終了の前に進捗を保存するために捕捉します
countdown=30	実行を N 秒遅らせます
eta=datetime(...)	指定した時刻に実行します

例： 高度なパターンについては、assets/celery_patterns.py を参照してください。

---

Celery の設定

設定	値	用途
BROKER_VISIBILITY_TIMEOUT	86400（24h）	長時間のタスクが再キューされるのを防ぎます
CELERY_RESULT_BACKEND	django-db	結果を PostgreSQL に保存します
CELERY_TASK_TRACK_STARTED	True	タスクの開始時刻を追跡します
soft_time_limit	タスクごとに設定	SoftTimeLimitExceeded を発生させます
time_limit	タスクごとに設定	強制終了（SIGKILL）

設定全体： assets/celery_patterns.py および config/celery.py、config/settings/celery.py の実際のファイルを参照してください。

---

パーティション分割テーブル向けの UUIDv7

Finding と ResourceFindingMapping は、時刻ベースのパーティション分割に UUIDv7 を使用します。

from uuid6 import uuid7
from api.uuid_utils import uuid7_start, uuid7_end, datetime_to_uuid7

# Partition-aware filtering
start = uuid7_start(datetime_to_uuid7(date_from))
end = uuid7_end(datetime_to_uuid7(date_to), settings.FINDINGS_TABLE_PARTITION_MONTHS)
queryset.filter(id__gte=start, id__lt=end)

UUIDv7 を使う理由： 時系列順の UUID を使用すると、範囲クエリの際に PostgreSQL がパーティションをプルーニングできます。

---

RLS を使用した一括操作

from api.db_utils import batch_delete, create_objects_in_batches, update_objects_in_batches

# Delete in batches (RLS-aware)
batch_delete(tenant_id, queryset, batch_size=1000)

# Bulk create with RLS
create_objects_in_batches(tenant_id, Finding, objects, batch_size=500)

# Bulk update with RLS
update_objects_in_batches(tenant_id, Finding, objects, fields=["status"], batch_size=500)

---

セキュリティパターン

完全な例：assets/security_patterns.py を参照してください。

テナント分離の概要

パターン	ルール
ViewSet での RLS	BaseRLSViewSet によって自動適用されます — tenant_id は JWT から取得します
Celery での RLS	@set_tenant と rls_transaction(tenant_id) を必ず使用します
テナントをまたぐ検証	多層防御：obj.tenant_id == request.tenant_id を確認します
ユーザー入力を決して信用しない	JWT から取得した request.tenant_id を使用し、request.data.get("tenant_id") は決して使用しません
管理者 DB のバイパス	テナントをまたぐ管理者操作に限って使用します — すべてのテナントのデータが公開されます

Celery タスクのセキュリティ概要

パターン	ルール
名前付きタスクのみ	ユーザー入力から動的にタスク名を指定しないでください
引数を検証する	データベースを検索する前に UUID の形式を確認してください
安全なキューイング	コミット後にキューへ追加するため、transaction.on_commit() を使用してください
最新のリトライ方式	autoretry_for、retry_backoff、retry_jitter を使用してください
時間制限	タスクの停止を防ぐため、soft_time_limit と time_limit を設定してください
冪等性	update_or_create または冪等性キーを使用してください

クイックリファレンス

# Safe task queuing - task only enqueued after transaction commits
with transaction.atomic():
    provider = Provider.objects.create(**data)
    transaction.on_commit(
        lambda: verify_provider_connection.delay(
            tenant_id=str(request.tenant_id),
            provider_id=str(provider.id)
        )
    )

# Modern retry pattern
@shared_task(
    base=RLSTask,
    bind=True,
    autoretry_for=(ConnectionError, TimeoutError, OperationalError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    max_retries=5,
    soft_time_limit=300,
    time_limit=360,
)
@set_tenant
def sync_provider_data(self, tenant_id, provider_id):
    with rls_transaction(tenant_id):
        # ... task logic
        pass

# Idempotent task - safe to retry
@shared_task(base=RLSTask, acks_late=True)
@set_tenant
def process_finding(tenant_id, finding_uid, data):
    with rls_transaction(tenant_id):
        Finding.objects.update_or_create(uid=finding_uid, defaults=data)

---

本番環境へのデプロイチェックリスト

すべての設定： references/production-settings.md を参照してください

本番環境へデプロイする前に、毎回実行してください。

cd api && uv run python src/backend/manage.py check --deploy

重要な設定

設定	本番環境での値	誤設定時のリスク
DEBUG	False	スタックトレース、設定、SQL クエリが漏えいします
SECRET_KEY	環境変数、ローテーション済み	セッションハイジャック、CSRF の回避
ALLOWED_HOSTS	明示的なリスト	Host ヘッダー攻撃
SECURE_SSL_REDIRECT	True	認証情報が HTTP 経由で送信されます
SESSION_COOKIE_SECURE	True	セッション Cookie が HTTP 経由で送信されます
CSRF_COOKIE_SECURE	True	CSRF トークンが HTTP 経由で送信されます
SECURE_HSTS_SECONDS	31536000（1 年）	ダウングレード攻撃
CONN_MAX_AGE	60 以上	接続プールの枯渇

---

コマンド

# Development
cd api && uv run python src/backend/manage.py runserver
cd api && uv run python src/backend/manage.py shell

# Celery
cd api && uv run celery -A config.celery worker -l info -Q scans,overview
cd api && uv run celery -A config.celery beat -l info

# Testing
cd api && uv run pytest -x --tb=short

# Production checks
cd api && uv run python src/backend/manage.py check --deploy

---

リソース

ローカルリファレンス

• ファイルの場所： references/file-locations.md を参照してください
• モデリングの決定事項： references/modeling-decisions.md を参照してください
• 設定： references/configuration.md を参照してください
• 本番環境の設定： references/production-settings.md を参照してください
• セキュリティパターン： assets/security_patterns.py を参照してください

関連スキル

• 汎用 DRF パターン：django-drf スキルを使用してください
• API テスト：prowler-test-api スキルを使用してください

Context7 MCP（推奨）

前提条件： 最新のドキュメントを検索できるよう、Context7 MCP サーバーをインストールしてください。

Prowler 固有のパターンを実装またはデバッグする際は、mcp_context7_query-docs を使って次のライブラリを検索してください。

ライブラリ	Context7 ID	用途
Celery	/websites/celeryq_dev_en_stable	タスクのパターン、キュー、エラー処理
django-celery-beat	/celery/django-celery-beat	定期タスクのスケジューリング
Django	/websites/djangoproject_en_5_2	モデル、ORM、制約、インデックス

クエリの例：

mcp_context7_query-docs(libraryId="/websites/celeryq_dev_en_stable", query="shared_task decorator retry patterns")
mcp_context7_query-docs(libraryId="/celery/django-celery-beat", query="periodic task database scheduler")
mcp_context7_query-docs(libraryId="/websites/djangoproject_en_5_2", query="model constraints CheckConstraint UniqueConstraint")

注： 正しいライブラリ ID を探す必要がある場合は、最初に mcp_context7_resolve-library-id を使用してください。