restricted_action_read_only_channel: the channel is locked

"""Work out whether a Slack channel will accept a message from this app.

Read only, and inferential on purpose: whether an app may post is the one thing
here that could be settled by posting, and posting is exactly what this section
does not do. The check reads the flag if the plan returns it, reads who has
actually spoken if it does not, and looks for this app among the authors.
"""
import argparse
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("slack_channel_posting_policy")

API = "https://slack.com/api/"

# Message subtypes anybody can generate, including in a channel nobody may post
# in. Joining a locked channel still writes a channel_join message, so counting
# these turns a silent room into a busy one and hides the finding.
NOT_A_POST = frozenset({
    "channel_join", "channel_leave", "channel_topic", "channel_purpose",
    "channel_name", "channel_archive", "channel_unarchive", "pinned_item",
    "unpinned_item", "bot_add", "bot_remove", "reminder_add", "tombstone",
    "channel_convert_to_private", "channel_convert_to_public",
})

# A channel with more members than this, and no more authors than that, is the
# shape of an announcement channel. Both numbers are judgement calls and both
# are meant to be edited for your workspace.
CROWD = 25
NARROW_AUTHORS = 3
MIN_SAMPLE = 8

REPAIR_ASK = ("ask a channel manager to open the channel, then Settings, "
              "Permissions, Posting permissions, and add this app to the members "
              "who may post")
REPAIR_MOVE = ("or point the integration at a channel without the restriction. An "
               "announcement channel is restricted deliberately and an app is not "
               "usually the exception people want to make")
REPAIR_NOT_SCOPES = ("do not add scopes for this. A posting permission is a "
                     "channel setting, and chat:write is already the whole grant "
                     "Slack has to offer")


def read_only_state(channel):
    """Report the flag and whether the flag was there at all. Pure.

    The distinction this function exists to preserve: channel.get("is_read_only")
    returns None on every plan that does not expose the field, None is falsey, and
    a check written that way reports every locked channel in the workspace as
    writable with complete confidence.
    """
    fields = channel or {}
    if "is_read_only" not in fields:
        return ("not-exposed",
                "conversations.info did not return is_read_only at all. Absent is "
                "not false: the field is only present on the plans that expose it, "
                "so this says nothing either way and the history below is the only "
                "evidence available.")
    if fields.get("is_read_only") is True:
        return ("declared-read-only",
                "conversations.info returns is_read_only: true. Nobody posts here, "
                "the app included, and no scope change alters that.")
    return ("declared-writable",
            "is_read_only came back false. That covers the whole-channel lock and "
            "says nothing about posting being restricted to a named group of "
            "members, which is the more common arrangement and sets no readable "
            "field.")


def authorship(messages, bot_user_id=None):
    """Who has actually posted in the readable history. Pure.

    Returns a stats dict. System subtypes are dropped before anybody is counted,
    because they are written on behalf of people who may hold no posting rights
    at all.
    """
    stats = {"sample": len(messages or []), "posts": 0, "apps": 0,
             "authors": [], "self_posted": False}
    for message in messages or []:
        subtype = message.get("subtype")
        if subtype in NOT_A_POST:
            continue
        author = (message.get("user") or message.get("bot_id")
                  or (message.get("bot_profile") or {}).get("id"))
        if not author:
            continue
        stats["posts"] += 1
        if subtype == "bot_message" or message.get("bot_id"):
            stats["apps"] += 1
        if bot_user_id and author == bot_user_id:
            stats["self_posted"] = True
        if author not in stats["authors"]:
            stats["authors"].append(author)
    return stats


def posting_verdict(state, stats, member_count):
    """Combine the flag, the authorship and the size of the room. Pure.

    Returns (verdict, detail). Only `locked` is a fact. `announcement-shaped` is
    an inference and says so in its own text, because the alternative explanation
    for three authors and nine hundred members is a channel nobody uses.
    """
    if state == "declared-read-only":
        return ("locked",
                "the channel declares itself read only. This is the one certain "
                "answer available to a read-only token, and it is an admin action "
                "to change.")

    posts = int(stats.get("posts") or 0)
    authors = list(stats.get("authors") or [])
    members = int(member_count or 0)

    if posts < MIN_SAMPLE:
        return ("undetermined",
                "only %d readable post(s) in the sample. Too little to infer "
                "anything from, and a small sample is itself worth knowing: a "
                "non-Marketplace app is clamped to 15 objects per "
                "conversations.history call." % posts)
    if stats.get("self_posted"):
        return ("app-has-posted",
                "this app is among the authors of the readable history, so it was "
                "permitted to post here at least once without anybody sending a "
                "test message to find out. Evidence about the past: a permission "
                "changed this morning appears in the next failed send, not here.")
    if len(authors) <= NARROW_AUTHORS and members >= CROWD:
        return ("announcement-shaped",
                "%d member(s) and %d distinct author(s) across %d readable "
                "post(s). That is the shape of a channel where posting is held by "
                "a few people. It is also the shape of a channel nobody talks in, "
                "so treat this as a question for a channel manager rather than as "
                "a finding." % (members, len(authors), posts))
    return ("open",
            "%d distinct author(s) across %d readable post(s) in a channel of %d. "
            "Nothing here looks like a posting restriction, and the app has simply "
            "not posted yet." % (len(authors), posts, members))


def whoami(session):
    """The bot's own user ID. GET only."""
    body = session.get(API + "auth.test", timeout=30).json()
    if body.get("ok") is not True:
        raise SystemExit("auth.test answered 200 with ok: false, error=%s"
                         % body.get("error"))
    return body.get("user_id"), body.get("team")


def channel_info(session, cid):
    """One channel object with its member count. GET only."""
    body = session.get(API + "conversations.info",
                       params={"channel": cid, "include_num_members": "true"},
                       timeout=30).json()
    if body.get("ok") is not True:
        raise SystemExit("conversations.info on %s answered ok: false, error=%s"
                         % (cid, body.get("error")))
    return body.get("channel") or {}


def recent_messages(session, cid, limit):
    """The readable history, or an honest empty list. GET only."""
    body = session.get(API + "conversations.history",
                       params={"channel": cid, "limit": limit}, timeout=30).json()
    if body.get("ok") is not True:
        log.warning("conversations.history on %s answered ok: false, error=%s. "
                    "The inference below has nothing to work from.",
                    cid, body.get("error"))
        return []
    return body.get("messages") or []


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("channels", nargs="+", help="channel IDs the integration posts to")
    ap.add_argument("--token-env", default="SLACK_BOT_TOKEN",
                    help="environment variable holding the bot token")
    ap.add_argument("--sample", type=int, default=100,
                    help="how many messages of history to read per channel")
    args = ap.parse_args()

    token = os.environ.get(args.token_env)
    if not token:
        log.error("set %s (channels:read and channels:history are enough)",
                  args.token_env)
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + token})

    me, team = whoami(s)
    log.info("identity  %s in %s", me, team)

    restricted = 0
    for cid in args.channels:
        channel = channel_info(s, cid)
        state, why = read_only_state(channel)
        members = channel.get("num_members") or 0
        stats = authorship(recent_messages(s, cid, args.sample), me)
        verdict, detail = posting_verdict(state, stats, members)

        log.info("%-12s flag     %-18s %s", cid, state, why)
        log.info("%-12s history  %d post(s) by %d author(s), channel has %s member(s)",
                 cid, stats["posts"], len(stats["authors"]), members)

        if verdict in ("app-has-posted", "open"):
            log.info("%-12s posting  %-18s %s", cid, verdict, detail)
            continue

        restricted += 1
        log.warning("%-12s posting  %-18s %s", cid, verdict, detail)
        log.warning("  repair: %s", REPAIR_ASK)
        log.warning("  repair: %s", REPAIR_MOVE)
        log.warning("  note:   %s", REPAIR_NOT_SCOPES)

    log.info("%d channel(s) checked, %d where posting looks restricted",
             len(args.channels), restricted)
    return 1 if restricted else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Work out whether a Slack channel will accept a message from this app.
 *
 * Read only, and inferential on purpose: whether an app may post is the one
 * thing here that could be settled by posting, and posting is exactly what this
 * section does not do. The check reads the flag if the plan returns it, reads
 * who has actually spoken if it does not, and looks for this app among the
 * authors.
 */

const API = 'https://slack.com/api/';

// Message subtypes anybody can generate, including in a channel nobody may post
// in. Joining a locked channel still writes a channel_join message, so counting
// these turns a silent room into a busy one and hides the finding.
const NOT_A_POST = new Set([
  'channel_join', 'channel_leave', 'channel_topic', 'channel_purpose',
  'channel_name', 'channel_archive', 'channel_unarchive', 'pinned_item',
  'unpinned_item', 'bot_add', 'bot_remove', 'reminder_add', 'tombstone',
  'channel_convert_to_private', 'channel_convert_to_public',
]);

// Both numbers are judgement calls and both are meant to be edited.
const CROWD = 25;
const NARROW_AUTHORS = 3;
const MIN_SAMPLE = 8;

const REPAIR_ASK = 'ask a channel manager to open the channel, then Settings, ' +
  'Permissions, Posting permissions, and add this app to the members who may post';
const REPAIR_MOVE = 'or point the integration at a channel without the ' +
  'restriction. An announcement channel is restricted deliberately and an app is ' +
  'not usually the exception people want to make';
const REPAIR_NOT_SCOPES = 'do not add scopes for this. A posting permission is a ' +
  'channel setting, and chat:write is already the whole grant Slack has to offer';

/**
 * Report the flag and whether the flag was there at all. Pure.
 * channel.is_read_only is undefined on every plan that does not expose it, and
 * undefined is falsey, so a check written the obvious way reports every locked
 * channel as writable.
 */
export function readOnlyState(channel) {
  const fields = channel ?? {};
  if (!Object.prototype.hasOwnProperty.call(fields, 'is_read_only')) {
    return ['not-exposed',
      'conversations.info did not return is_read_only at all. Absent is not ' +
      'false: the field is only present on the plans that expose it, so this says ' +
      'nothing either way and the history below is the only evidence available.'];
  }
  if (fields.is_read_only === true) {
    return ['declared-read-only',
      'conversations.info returns is_read_only: true. Nobody posts here, the app ' +
      'included, and no scope change alters that.'];
  }
  return ['declared-writable',
    'is_read_only came back false. That covers the whole-channel lock and says ' +
    'nothing about posting being restricted to a named group of members, which is ' +
    'the more common arrangement and sets no readable field.'];
}

/**
 * Who has actually posted in the readable history. Pure.
 * System subtypes are dropped before anybody is counted.
 */
export function authorship(messages, botUserId = null) {
  const stats = {
    sample: (messages ?? []).length, posts: 0, apps: 0,
    authors: [], self_posted: false,
  };
  for (const message of messages ?? []) {
    const subtype = message.subtype;
    if (NOT_A_POST.has(subtype)) continue;
    const author = message.user ?? message.bot_id ?? message.bot_profile?.id ?? null;
    if (!author) continue;
    stats.posts += 1;
    if (subtype === 'bot_message' || message.bot_id) stats.apps += 1;
    if (botUserId && author === botUserId) stats.self_posted = true;
    if (!stats.authors.includes(author)) stats.authors.push(author);
  }
  return stats;
}

/**
 * Combine the flag, the authorship and the size of the room. Pure.
 * Only `locked` is a fact. `announcement-shaped` is an inference and says so.
 */
export function postingVerdict(state, stats, memberCount) {
  if (state === 'declared-read-only') {
    return ['locked',
      'the channel declares itself read only. This is the one certain answer ' +
      'available to a read-only token, and it is an admin action to change.'];
  }

  const posts = Number(stats?.posts ?? 0);
  const authors = stats?.authors ?? [];
  const members = Number(memberCount ?? 0);

  if (posts < MIN_SAMPLE) {
    return ['undetermined',
      `only ${posts} readable post(s) in the sample. Too little to infer anything ` +
      'from, and a small sample is itself worth knowing: a non-Marketplace app is ' +
      'clamped to 15 objects per conversations.history call.'];
  }
  if (stats.self_posted) {
    return ['app-has-posted',
      'this app is among the authors of the readable history, so it was permitted ' +
      'to post here at least once without anybody sending a test message to find ' +
      'out. Evidence about the past: a permission changed this morning appears in ' +
      'the next failed send, not here.'];
  }
  if (authors.length <= NARROW_AUTHORS && members >= CROWD) {
    return ['announcement-shaped',
      `${members} member(s) and ${authors.length} distinct author(s) across ` +
      `${posts} readable post(s). That is the shape of a channel where posting is ` +
      'held by a few people. It is also the shape of a channel nobody talks in, so ' +
      'treat this as a question for a channel manager rather than as a finding.'];
  }
  return ['open',
    `${authors.length} distinct author(s) across ${posts} readable post(s) in a ` +
    `channel of ${members}. Nothing here looks like a posting restriction, and the ` +
    'app has simply not posted yet.'];
}

async function get(token, method, params) {
  const res = await fetch(`${API}${method}?${new URLSearchParams(params)}`, {
    headers: { Authorization: `Bearer ${token}` },
  });
  return res.json();
}

function arg(args, name, fallback = null) {
  const i = args.indexOf(name);
  return i === -1 ? fallback : args[i + 1];
}

function positionals(args) {
  const out = [];
  for (let i = 0; i < args.length; i += 1) {
    if (args[i].startsWith('--')) { i += 1; continue; }
    out.push(args[i]);
  }
  return out;
}

async function main() {
  const args = process.argv.slice(2);
  const channels = positionals(args);
  if (channels.length === 0) {
    console.error('usage: <channel id>... [--token-env SLACK_BOT_TOKEN] [--sample 100]');
    process.exitCode = 2;
    return;
  }
  const tokenEnv = arg(args, '--token-env', 'SLACK_BOT_TOKEN');
  const token = process.env[tokenEnv];
  if (!token) {
    console.error(`set ${tokenEnv} (channels:read and channels:history are enough)`);
    process.exitCode = 2;
    return;
  }
  const sample = Number(arg(args, '--sample', 100));

  const auth = await get(token, 'auth.test', {});
  if (auth.ok !== true) {
    throw new Error(`auth.test answered 200 with ok: false, error=${auth.error}`);
  }
  console.log(`identity  ${auth.user_id} in ${auth.team}`);

  let restricted = 0;
  for (const cid of channels) {
    const info = await get(token, 'conversations.info',
      { channel: cid, include_num_members: 'true' });
    if (info.ok !== true) {
      throw new Error(`conversations.info on ${cid} answered ok: false, error=${info.error}`);
    }
    const channel = info.channel ?? {};
    const [state, why] = readOnlyState(channel);
    const members = channel.num_members ?? 0;

    const history = await get(token, 'conversations.history',
      { channel: cid, limit: String(sample) });
    if (history.ok !== true) {
      console.warn(`conversations.history on ${cid} answered ok: false, error=` +
        `${history.error}. The inference below has nothing to work from.`);
    }
    const stats = authorship(history.messages ?? [], auth.user_id);
    const [verdict, detail] = postingVerdict(state, stats, members);

    console.log(`${cid.padEnd(12)} flag     ${state.padEnd(18)} ${why}`);
    console.log(`${cid.padEnd(12)} history  ${stats.posts} post(s) by ` +
      `${stats.authors.length} author(s), channel has ${members} member(s)`);

    if (verdict === 'app-has-posted' || verdict === 'open') {
      console.log(`${cid.padEnd(12)} posting  ${verdict.padEnd(18)} ${detail}`);
      continue;
    }

    restricted += 1;
    console.warn(`${cid.padEnd(12)} posting  ${verdict.padEnd(18)} ${detail}`);
    console.warn(`  repair: ${REPAIR_ASK}`);
    console.warn(`  repair: ${REPAIR_MOVE}`);
    console.warn(`  note:   ${REPAIR_NOT_SCOPES}`);
  }

  console.log(`${channels.length} channel(s) checked, ${restricted} where posting ` +
    'looks restricted');
  process.exitCode = restricted ? 1 : 0;
}

// Only run when invoked directly, so importing this module in the tests does not
// execute main() and fail the file on a missing token.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from slack_channel_posting_policy import (authorship, posting_verdict,
                                           read_only_state)

APP = "U0APPBOT11"


def crowd(authors, count):
    """A readable history with `count` posts spread over `authors`."""
    return [{"user": authors[i % len(authors)], "ts": "170000000%d.0001" % i}
            for i in range(count)]


def test_a_missing_flag_is_not_a_writable_channel():
    state, why = read_only_state({"id": "C0LOCKED11", "name": "announce"})
    assert state == "not-exposed"
    assert "Absent is not false" in why


def test_the_flag_is_reported_when_the_plan_returns_it():
    assert read_only_state({"is_read_only": True})[0] == "declared-read-only"
    assert read_only_state({"is_read_only": False})[0] == "declared-writable"


def test_a_declared_read_only_channel_is_the_one_certain_verdict():
    verdict, detail = posting_verdict("declared-read-only", {"posts": 0}, 900)
    assert verdict == "locked"
    assert "admin action" in detail


def test_joins_and_topic_changes_are_not_posts():
    messages = [{"user": "U0AAA", "subtype": "channel_join"},
                {"user": "U0BBB", "subtype": "channel_topic"},
                {"user": "U0CCC", "subtype": "pinned_item"}]
    stats = authorship(messages, APP)
    assert stats["sample"] == 3
    assert stats["posts"] == 0
    assert stats["authors"] == []


def test_authors_are_counted_once_each():
    stats = authorship(crowd(["U0AAA", "U0BBB"], 10), APP)
    assert stats["posts"] == 10
    assert stats["authors"] == ["U0AAA", "U0BBB"]
    assert stats["self_posted"] is False


def test_the_app_finds_itself_in_the_history():
    messages = crowd(["U0AAA"], 9) + [{"user": APP, "bot_id": "B0APP1",
                                       "subtype": "bot_message"}]
    stats = authorship(messages, APP)
    assert stats["self_posted"] is True
    assert stats["apps"] == 1
    verdict, detail = posting_verdict("not-exposed", stats, 900)
    assert verdict == "app-has-posted"
    assert "past" in detail


def test_an_app_message_without_a_user_field_still_has_an_author():
    stats = authorship([{"bot_id": "B0OTHER1", "subtype": "bot_message"}], APP)
    assert stats["posts"] == 1
    assert stats["authors"] == ["B0OTHER1"]
    assert stats["self_posted"] is False


def test_a_crowd_with_three_voices_is_announcement_shaped():
    stats = authorship(crowd(["U0AAA", "U0BBB", "U0CCC"], 60), APP)
    verdict, detail = posting_verdict("not-exposed", stats, 900)
    assert verdict == "announcement-shaped"
    assert "channel manager" in detail


def test_three_voices_in_a_small_room_are_just_a_small_team():
    stats = authorship(crowd(["U0AAA", "U0BBB", "U0CCC"], 60), APP)
    assert posting_verdict("not-exposed", stats, 9)[0] == "open"


def test_a_thin_sample_produces_no_verdict_at_all():
    stats = authorship(crowd(["U0AAA"], 4), APP)
    verdict, detail = posting_verdict("declared-writable", stats, 900)
    assert verdict == "undetermined"
    assert "15 objects" in detail


def test_the_flag_outranks_the_inference():
    stats = authorship(crowd(["U0AAA", "U0BBB", "U0CCC", "U0DDD"], 40), APP)
    assert posting_verdict("declared-read-only", stats, 40)[0] == "locked"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { authorship, postingVerdict, readOnlyState }
  from './slack-channel-posting-policy.mjs';

const APP = 'U0APPBOT11';

/** A readable history with `count` posts spread over `authors`. */
function crowd(authors, count) {
  return Array.from({ length: count }, (_, i) => ({
    user: authors[i % authors.length], ts: `170000000${i}.0001`,
  }));
}

test('a missing flag is not a writable channel', () => {
  const [state, why] = readOnlyState({ id: 'C0LOCKED11', name: 'announce' });
  assert.equal(state, 'not-exposed');
  assert.match(why, /Absent is not false/);
});

test('the flag is reported when the plan returns it', () => {
  assert.equal(readOnlyState({ is_read_only: true })[0], 'declared-read-only');
  assert.equal(readOnlyState({ is_read_only: false })[0], 'declared-writable');
});

test('a declared read only channel is the one certain verdict', () => {
  const [verdict, detail] = postingVerdict('declared-read-only', { posts: 0 }, 900);
  assert.equal(verdict, 'locked');
  assert.match(detail, /admin action/);
});

test('joins and topic changes are not posts', () => {
  const stats = authorship([
    { user: 'U0AAA', subtype: 'channel_join' },
    { user: 'U0BBB', subtype: 'channel_topic' },
    { user: 'U0CCC', subtype: 'pinned_item' },
  ], APP);
  assert.equal(stats.sample, 3);
  assert.equal(stats.posts, 0);
  assert.deepEqual(stats.authors, []);
});

test('authors are counted once each', () => {
  const stats = authorship(crowd(['U0AAA', 'U0BBB'], 10), APP);
  assert.equal(stats.posts, 10);
  assert.deepEqual(stats.authors, ['U0AAA', 'U0BBB']);
  assert.equal(stats.self_posted, false);
});

test('the app finds itself in the history', () => {
  const messages = [...crowd(['U0AAA'], 9),
    { user: APP, bot_id: 'B0APP1', subtype: 'bot_message' }];
  const stats = authorship(messages, APP);
  assert.equal(stats.self_posted, true);
  assert.equal(stats.apps, 1);
  const [verdict, detail] = postingVerdict('not-exposed', stats, 900);
  assert.equal(verdict, 'app-has-posted');
  assert.match(detail, /past/);
});

test('an app message without a user field still has an author', () => {
  const stats = authorship([{ bot_id: 'B0OTHER1', subtype: 'bot_message' }], APP);
  assert.equal(stats.posts, 1);
  assert.deepEqual(stats.authors, ['B0OTHER1']);
  assert.equal(stats.self_posted, false);
});

test('a crowd with three voices is announcement shaped', () => {
  const stats = authorship(crowd(['U0AAA', 'U0BBB', 'U0CCC'], 60), APP);
  const [verdict, detail] = postingVerdict('not-exposed', stats, 900);
  assert.equal(verdict, 'announcement-shaped');
  assert.match(detail, /channel manager/);
});

test('three voices in a small room are just a small team', () => {
  const stats = authorship(crowd(['U0AAA', 'U0BBB', 'U0CCC'], 60), APP);
  assert.equal(postingVerdict('not-exposed', stats, 9)[0], 'open');
});

test('a thin sample produces no verdict at all', () => {
  const stats = authorship(crowd(['U0AAA'], 4), APP);
  const [verdict, detail] = postingVerdict('declared-writable', stats, 900);
  assert.equal(verdict, 'undetermined');
  assert.match(detail, /15 objects/);
});

test('the flag outranks the inference', () => {
  const stats = authorship(crowd(['U0AAA', 'U0BBB', 'U0CCC', 'U0DDD'], 40), APP);
  assert.equal(postingVerdict('declared-read-only', stats, 40)[0], 'locked');
});