feat: channel manager api

db/capabilities.yml

@@ -0,0 +1,4 @@
+capabilities:
+  channel_manager:
+    label: Channel manager lookups
+    covers: Apps can ask whether you manage a given channel. They are told yes, no, or nothing.

db/flags.yml

@@ -11,3 +11,8 @@ flags:
     label: Decisions
     covers: The decisions section, and linking a case to one
     default: true
+  public_api:
+    label: Public API
+    covers: Channel manager lookups, tokens and opt-in
+    audience: Everyone in Slack
+    default: false

db/init.sql

@@ -2,6 +2,7 @@ CREATE SCHEMA IF NOT EXISTS raw;
 CREATE SCHEMA IF NOT EXISTS analytics;
 CREATE SCHEMA IF NOT EXISTS app;
 CREATE SCHEMA IF NOT EXISTS fd;
+CREATE SCHEMA IF NOT EXISTS api;
 
 DO $$
 DECLARE
@@ -71,6 +72,25 @@ BEGIN
     EXECUTE 'GRANT USAGE ON ALL SEQUENCES IN SCHEMA fd TO rails_app';
     EXECUTE 'ALTER DEFAULT PRIVILEGES IN SCHEMA fd GRANT USAGE ON SEQUENCES TO rails_app';
 
+    EXECUTE 'GRANT USAGE ON SCHEMA api TO rails_app';
+    EXECUTE 'GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA api TO rails_app';
+    EXECUTE 'ALTER DEFAULT PRIVILEGES IN SCHEMA api '
+        'GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO rails_app';
+    EXECUTE 'GRANT USAGE ON ALL SEQUENCES IN SCHEMA api TO rails_app';
+    EXECUTE 'ALTER DEFAULT PRIVILEGES IN SCHEMA api GRANT USAGE ON SEQUENCES TO rails_app';
+
+    EXECUTE 'GRANT USAGE ON SCHEMA api TO pipeline_writer';
+    EXECUTE 'GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA api '
+        'TO pipeline_writer';
+    EXECUTE 'ALTER DEFAULT PRIVILEGES IN SCHEMA api '
+        'GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO pipeline_writer';
+    EXECUTE 'GRANT USAGE ON ALL SEQUENCES IN SCHEMA api TO pipeline_writer';
+    EXECUTE 'ALTER DEFAULT PRIVILEGES IN SCHEMA api '
+        'GRANT USAGE ON SEQUENCES TO pipeline_writer';
+
+    EXECUTE 'REVOKE ALL ON ALL TABLES IN SCHEMA api FROM dbt_owner';
+    EXECUTE 'REVOKE ALL ON SCHEMA api FROM dbt_owner';
+
     IF EXISTS (SELECT 1 FROM pg_tables WHERE schemaname = 'app' AND tablename = 'sync_request') THEN
         EXECUTE 'GRANT SELECT, UPDATE ON app.sync_request TO pipeline_writer';
     END IF;

db/migrations/0055_api_and_channel_managers.sql

@@ -0,0 +1,82 @@
+CREATE SCHEMA IF NOT EXISTS api;
+
+CREATE TABLE api.channel_manager (
+    channel_id  text NOT NULL,
+    user_id     text NOT NULL,
+    assigned_at timestamptz,
+    PRIMARY KEY (channel_id, user_id)
+);
+
+CREATE INDEX channel_manager_user_idx ON api.channel_manager (user_id);
+
+CREATE TABLE api.channel_sweep (
+    channel_id text PRIMARY KEY,
+    synced_at  timestamptz NOT NULL DEFAULT now(),
+    managers   integer NOT NULL DEFAULT 0
+);
+
+CREATE TABLE api.consent (
+    user_id          text NOT NULL,
+    capability       text NOT NULL,
+    state            text NOT NULL,
+    changed_at       timestamptz NOT NULL DEFAULT now(),
+    changed_via      text NOT NULL,
+    first_granted_at timestamptz,
+    PRIMARY KEY (user_id, capability),
+    CONSTRAINT consent_state CHECK (state IN ('granted', 'withheld'))
+);
+
+CREATE INDEX consent_granted_idx ON api.consent (capability)
+    WHERE state = 'granted';
+
+CREATE TABLE api.consent_log (
+    id         bigserial PRIMARY KEY,
+    user_id    text NOT NULL,
+    capability text NOT NULL,
+    state      text NOT NULL,
+    via        text NOT NULL,
+    at         timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE INDEX consent_log_at_idx ON api.consent_log (at DESC);
+
+CREATE TABLE api.token (
+    id            bigserial PRIMARY KEY,
+    owner_user_id text NOT NULL,
+    name          text NOT NULL,
+    prefix        text NOT NULL,
+    digest        text NOT NULL UNIQUE,
+    rate_limit    integer,
+    created_at    timestamptz NOT NULL DEFAULT now(),
+    last_used_at  timestamptz,
+    revoked_at    timestamptz,
+    revoked_by    text,
+    CONSTRAINT token_rate_limit_sane CHECK (rate_limit IS NULL OR rate_limit > 0)
+);
+
+CREATE INDEX token_owner_idx ON api.token (owner_user_id)
+    WHERE revoked_at IS NULL;
+
+CREATE TABLE api.setting (
+    key        text PRIMARY KEY,
+    value      integer NOT NULL,
+    changed_by text,
+    changed_at timestamptz NOT NULL DEFAULT now()
+);
+
+INSERT INTO api.setting (key, value) VALUES
+    ('rate_per_minute', 100),
+    ('batch_max', 100),
+    ('tokens_per_owner', 3);
+
+CREATE TABLE api.request_log (
+    id              bigserial PRIMARY KEY,
+    token_id        bigint NOT NULL REFERENCES api.token (id),
+    channel_id      text,
+    subject_user_id text,
+    outcome         text NOT NULL,
+    at              timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE INDEX request_log_subject_idx ON api.request_log (subject_user_id, at DESC);
+CREATE INDEX request_log_token_idx ON api.request_log (token_id, at DESC);

db/migrations/0056_api_event_log.sql

@@ -0,0 +1,10 @@
+CREATE TABLE api.event_log (
+    id            bigserial PRIMARY KEY,
+    actor_user_id text,
+    verb          text NOT NULL,
+    subject       text,
+    detail        text,
+    at            timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE INDEX event_log_at_idx ON api.event_log (at DESC);

db/migrations/0057_api_rate_default.sql

@@ -0,0 +1,4 @@
+UPDATE api.setting
+SET value = 20
+WHERE key = 'rate_per_minute'
+  AND value = 100;

db/migrations/0058_api_drop_batch_max.sql

@@ -0,0 +1 @@
+DELETE FROM api.setting WHERE key = 'batch_max';

db/migrations/0059_api_token_expiry.sql

@@ -0,0 +1,4 @@
+ALTER TABLE api.token ADD COLUMN expires_at timestamptz;
+
+CREATE INDEX token_expiry_idx ON api.token (expires_at)
+    WHERE revoked_at IS NULL AND expires_at IS NOT NULL;

deploy/env/serve.env.example

@@ -28,6 +28,7 @@ FD_ENCRYPTION_DETERMINISTIC_KEY=
 FD_ENCRYPTION_SALT=
 INTERNAL_PROXY_URL=
 SLACKSCAN_URL=
+SLACK_CHANNEL_MANAGER_ROLE_ID=
 PROXY_TOKEN_WEB=
 PROXY_ALLOW_PLAINTEXT=
 RAILS_MAX_THREADS=5

pipeline/checks/slack_roles.py

@@ -0,0 +1,168 @@
+import argparse
+import os
+import sys
+from collections import defaultdict
+
+from dotenv import load_dotenv
+from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError
+
+from lib.paths import ENV_FILE
+
+PAGE_SIZE = 200
+SAMPLES = 3
+KINDS = {"C": "channel", "G": "private group", "T": "workspace", "E": "enterprise"}
+
+BLOCKING = {
+    "missing_scope",
+    "not_allowed_token_type",
+    "invalid_auth",
+    "account_inactive",
+    "token_revoked",
+    "not_authed",
+}
+
+
+def die(message):
+    sys.exit(f"slack_roles: {message}")
+
+
+def api(token):
+    if not token:
+        die("no token. pass one as an argument or set SLACK_TOKEN to an xoxp user token")
+    return WebClient(token=token)
+
+
+def whoami(client):
+    try:
+        said = client.auth_test().data
+    except SlackApiError as exc:
+        die(f"auth.test failed: {exc.response.get('error', 'unknown_error')}")
+    return said
+
+
+def call(client, **params):
+    try:
+        return client.admin_roles_listAssignments(limit=PAGE_SIZE, **params)
+    except SlackApiError as exc:
+        error = exc.response.get("error", "unknown_error")
+        if error == "missing_scope":
+            needed = exc.response.get("needed", "admin.roles:read")
+            die(f"the token is missing {needed}. add it and reinstall the app")
+        if error in BLOCKING:
+            die(f"admin.roles.listAssignments refused this token: {error}")
+        die(f"admin.roles.listAssignments failed: {error}")
+
+
+def walk(client, pages=None, **params):
+    cursor = ""
+    seen = 0
+    while True:
+        page = call(client, cursor=cursor or None, **params)
+        for one in page.get("role_assignments", []):
+            yield one
+        seen += 1
+        cursor = page.get("response_metadata", {}).get("next_cursor") or ""
+        if not cursor or (pages is not None and seen >= pages):
+            return
+
+
+def kind_of(entity_id):
+    return KINDS.get(entity_id[:1], entity_id[:1] or "?")
+
+
+def gather(client, pages):
+    found = defaultdict(lambda: {"count": 0, "entities": set(), "kinds": set(), "users": set()})
+    for one in walk(client, pages=pages):
+        entity = one.get("entity_id", "")
+        role = found[one.get("role_id", "?")]
+        role["count"] += 1
+        role["entities"].add(entity)
+        role["kinds"].add(kind_of(entity))
+        role["users"].add(one.get("user_id", ""))
+    return found
+
+
+def report(found, capped):
+    print()
+    print(f"  {'role_id':<10}{'assigned':>10}{'entities':>10}  {'kind':<14}samples")
+    for role_id, one in sorted(found.items(), key=lambda pair: -pair[1]["count"]):
+        kinds = ", ".join(sorted(one["kinds"]))
+        samples = ", ".join(sorted(one["entities"])[:SAMPLES])
+        print(f"  {role_id:<10}{one['count']:>10}{len(one['entities']):>10}  {kinds:<14}{samples}")
+    if capped:
+        print(f"\n  counts cover the first {capped} pages only. pass --all to sweep everything.")
+
+
+def channel_role(found):
+    named = [role for role, one in found.items() if "channel" in one["kinds"]]
+    if len(named) == 1:
+        return named[0]
+    if not named:
+        print("\n  no role has channel entities. nothing here maps to channel manager.")
+        return None
+    print(f"\n  more than one role has channel entities: {', '.join(sorted(named))}")
+    print("  pass --role to check one of them against a channel you know.")
+    return None
+
+
+def check_filter(client, role_id, channel_id):
+    print(f"\n  filter check: role_ids={role_id} entity_ids={channel_id}")
+    rows = list(walk(client, role_ids=[role_id], entity_ids=[channel_id]))
+    if not rows:
+        print("  nothing came back. the filter may not combine, or nobody manages that channel.")
+        return
+    stray = {row.get("entity_id") for row in rows} - {channel_id}
+    users = sorted({row.get("user_id", "") for row in rows})
+    print(f"  {len(rows)} assignments, {len(users)} people")
+    print(f"  managers: {', '.join(users[:8])}{' ...' if len(users) > 8 else ''}")
+    if stray:
+        print(f"  WARNING entity_ids did not narrow it. also returned: {', '.join(sorted(stray)[:5])}")
+    else:
+        print("  every row is on that channel, so one call answers one channel")
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        prog="slack_roles",
+        description="find which Slack role id means channel manager, and prove that "
+                    "role_ids and entity_ids together answer one channel in one call",
+    )
+    parser.add_argument("token", nargs="?", help="an xoxp user token, else SLACK_TOKEN is used")
+    parser.add_argument("--all", action="store_true", help="sweep every page instead of the first few")
+    parser.add_argument("--pages", type=int, default=10, help="pages to sample, 200 per page")
+    parser.add_argument("--role", help="skip discovery and check this role id")
+    parser.add_argument("--channel", help="channel id for the filter check")
+    args = parser.parse_args()
+
+    load_dotenv(ENV_FILE)
+    client = api(args.token or os.environ.get("SLACK_TOKEN", "").strip())
+    who = whoami(client)
+    print(f"admin.roles.listAssignments as {who.get('user')} ({who.get('user_id')}) "
+          f"on {who.get('team')} / {who.get('team_id')}")
+
+    role_id = args.role
+    channel_id = args.channel
+
+    if role_id is None:
+        pages = None if args.all else args.pages
+        found = gather(client, pages)
+        if not found:
+            die("no role assignments came back at all")
+        report(found, None if args.all else args.pages)
+        role_id = channel_role(found)
+        if role_id is None:
+            return
+        if channel_id is None:
+            channel_id = sorted(found[role_id]["entities"])[0]
+        print(f"\n  channel manager looks like {role_id}")
+        print(f"  pin it:  SLACK_CHANNEL_MANAGER_ROLE_ID={role_id}")
+
+    if channel_id is None:
+        die("pass --channel with --role so the filter can be checked")
+
+    check_filter(client, role_id, channel_id)
+
+
+if __name__ == "__main__":
+    main()

pipeline/lib/config.py

@@ -33,6 +33,7 @@
             "FD_ENCRYPTION_SALT",
             "INTERNAL_PROXY_URL",
             "SLACKSCAN_URL",
+            "SLACK_CHANNEL_MANAGER_ROLE_ID",
             "PROXY_TOKEN_WEB",
             "PROXY_ALLOW_PLAINTEXT",
             "RAILS_MAX_THREADS",

pipeline/slack.manifest.yaml

@@ -17,6 +17,7 @@ oauth_config:
       - admin.analytics:read
       - admin.users:read
       - admin.teams:read
+      - admin.roles:read
 
 settings:
   event_subscriptions:

proxy/app.py

@@ -29,6 +29,7 @@
     "admin": frozenset(
         {
             "admin.users.list",
+            "admin.roles.listAssignments",
             "search.messages",
             "conversations.replies",
         }
@@ -44,6 +45,11 @@
             "admin.analytics.getAvailableDateRange",
         }
     ),
+    "admin": frozenset(
+        {
+            "admin.roles.listAssignments",
+        }
+    ),
 }
 
 CREDENTIALS = ("internal", "admin")