files.upload is retired: one probe returns method_deprecated

"""Confirm whether files.upload is dead for this app, and whether it was noticed.

Read only. The probe calls files.upload with no arguments, which cannot create
anything: it exists to be refused, and the refusal is the finding. The migration
is printed, never performed.
"""
import argparse
import logging
import os
import sys
import time

import requests

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

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

# 12 November 2025, 00:00 UTC: the day files.upload was sunset for all apps.
# The date was announced for 11 March 2025 and moved once.
SUNSET = 1762905600

DEAD = {"method_deprecated", "deprecated_endpoint"}
# Errors that mean the method answered rather than refused to exist.
ALIVE = {"no_file_data", "no_file_or_content", "invalid_arguments", "posting_to_general_channel_denied"}


def verdict(body):
    """Classify the argument-free files.upload probe. Pure, so it runs offline.

    The probe's whole job is to distinguish "this method no longer exists" from
    "this method exists and you called it wrong", and both arrive as HTTP 200.
    """
    if not isinstance(body, dict):
        return ("unreadable",
                "the probe got a body that is not JSON, so something other than "
                "Slack answered. Nothing can be concluded about the method.")
    error = body.get("error")
    if body.get("ok") is True:
        return ("unexpected",
                "ok: true from a call with no file. Read the response by hand "
                "before trusting anything else here.")
    if error in DEAD:
        return ("retired",
                "files.upload answered %s. The method was sunset for all apps on "
                "2025-11-12 and will not come back." % error)
    if error == "missing_scope":
        return ("unknown",
                "missing_scope: needed=%s. The probe never reached the method, so "
                "this says nothing about whether it is alive. Migrate anyway."
                % (body.get("needed") or "?"))
    if error in ("invalid_auth", "not_authed", "token_revoked", "account_inactive"):
        return ("auth",
                "error=%s. That is the token, not the method. Fix the credential "
                "and re-run before concluding anything." % error)
    if error in ALIVE:
        return ("still-answering",
                "error=%s, which means the method parsed the call rather than "
                "refusing to exist. Unexpected after the sunset, and still not a "
                "reason to stay on it." % error)
    return ("other",
            "error=%s. Not a deprecation answer; read it before acting."
            % (error or "<no error field>"))


def upload_activity(files, now=None, sunset=SUNSET):
    """Classify this app's own upload history against the cutover. Pure.

    `files` is the files.list array, restricted to files this bot uploaded.
    A fleet that has been failing since the sunset has no files after it.
    """
    stamps = sorted(int(f.get("created") or 0) for f in files)
    if not stamps:
        return ("no-uploads",
                "this app has uploaded no files the token can see, so there is "
                "no history to date the breakage from.")
    newest = stamps[-1]
    after = [s for s in stamps if s >= sunset]
    if after:
        return ("uploading",
                "%d file(s) uploaded after the 2025-11-12 cutover, so some caller "
                "already speaks the replacement flow." % len(after))
    days = int(((now or time.time()) - newest) / 86400)
    return ("silent-since-sunset",
            "newest upload is %d day(s) old and predates the cutover. Every "
            "caller has been failing since, quietly, at HTTP 200." % days)


def get(session, method, **params):
    r = session.get(API + method, params=params, timeout=30)
    try:
        return r.json()
    except ValueError:
        return r.text


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--count", type=int, default=100,
                    help="how many of the app's own files to read (default 100)")
    args = ap.parse_args()

    token = os.environ.get("SLACK_BOT_TOKEN")
    if not token:
        log.error("set SLACK_BOT_TOKEN (files:read is enough for the corroboration)")
        return 2

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

    state, detail = verdict(get(s, "files.upload"))
    bad = 0
    if state == "retired":
        bad += 1
        log.warning("%-19s %s", state, detail)
        log.warning("  repair: files.getUploadURLExternal(filename, length) -> upload the "
                    "raw bytes to upload_url -> files.completeUploadExternal(files, channel_id)")
        log.warning("  or use the SDK helper: client.files_upload_v2(...) / "
                    "client.filesUploadV2({...})")
    elif state in ("still-answering", "unknown", "auth", "unreadable", "unexpected", "other"):
        log.warning("%-19s %s", state, detail)
    else:
        log.info("%-19s %s", state, detail)

    me = get(s, "auth.test")
    if isinstance(me, dict) and me.get("ok") is True:
        listing = get(s, "files.list", user=me.get("user_id"), count=str(args.count))
        if isinstance(listing, dict) and listing.get("ok") is True:
            hstate, hdetail = upload_activity(listing.get("files") or [])
            if hstate == "silent-since-sunset":
                bad += 1
                log.warning("%-19s %s", hstate, hdetail)
            else:
                log.info("%-19s %s", hstate, hdetail)
        else:
            log.info("%-19s files.list did not answer ok: true (%s); the probe "
                     "above stands on its own", "no-history",
                     isinstance(listing, dict) and listing.get("error") or "?")
    else:
        log.info("%-19s auth.test did not answer ok: true, so the history check "
                 "was skipped", "no-history")

    log.info("1 method probed, %d finding(s)", bad)
    return 1 if bad else 0


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

/**
 * Confirm whether files.upload is dead for this app, and whether it was noticed.
 *
 * Read only. The probe calls files.upload with no arguments, which cannot create
 * anything: it exists to be refused, and the refusal is the finding. The
 * migration is printed, never performed.
 */
const API = 'https://slack.com/api/';

// 12 November 2025, 00:00 UTC: the day files.upload was sunset for all apps.
// The date was announced for 11 March 2025 and moved once.
export const SUNSET = 1762905600;

const DEAD = new Set(['method_deprecated', 'deprecated_endpoint']);
// Errors that mean the method answered rather than refused to exist.
const ALIVE = new Set([
  'no_file_data', 'no_file_or_content', 'invalid_arguments',
  'posting_to_general_channel_denied',
]);

/**
 * Classify the argument-free files.upload probe. Pure, so it runs offline.
 * Its job is to separate "this method no longer exists" from "this method
 * exists and you called it wrong", both of which arrive as HTTP 200.
 */
export function verdict(body) {
  if (typeof body !== 'object' || body === null || Array.isArray(body)) {
    return ['unreadable',
      'the probe got a body that is not JSON, so something other than Slack ' +
      'answered. Nothing can be concluded about the method.'];
  }
  const error = body.error;
  if (body.ok === true) {
    return ['unexpected',
      'ok: true from a call with no file. Read the response by hand before ' +
      'trusting anything else here.'];
  }
  if (DEAD.has(error)) {
    return ['retired',
      `files.upload answered ${error}. The method was sunset for all apps on ` +
      '2025-11-12 and will not come back.'];
  }
  if (error === 'missing_scope') {
    return ['unknown',
      `missing_scope: needed=${body.needed ?? '?'}. The probe never reached the ` +
      'method, so this says nothing about whether it is alive. Migrate anyway.'];
  }
  if (['invalid_auth', 'not_authed', 'token_revoked', 'account_inactive'].includes(error)) {
    return ['auth',
      `error=${error}. That is the token, not the method. Fix the credential and ` +
      're-run before concluding anything.'];
  }
  if (ALIVE.has(error)) {
    return ['still-answering',
      `error=${error}, which means the method parsed the call rather than ` +
      'refusing to exist. Unexpected after the sunset, and still not a reason ' +
      'to stay on it.'];
  }
  return ['other',
    `error=${error ?? '<no error field>'}. Not a deprecation answer; read it ` +
    'before acting.'];
}

/**
 * Classify this app's own upload history against the cutover. Pure.
 * `files` is the files.list array restricted to files this bot uploaded.
 */
export function uploadActivity(files, now = null, sunset = SUNSET) {
  const stamps = files.map((f) => Number(f.created ?? 0)).sort((a, b) => a - b);
  if (!stamps.length) {
    return ['no-uploads',
      'this app has uploaded no files the token can see, so there is no history ' +
      'to date the breakage from.'];
  }
  const newest = stamps[stamps.length - 1];
  const after = stamps.filter((s) => s >= sunset);
  if (after.length) {
    return ['uploading',
      `${after.length} file(s) uploaded after the 2025-11-12 cutover, so some ` +
      'caller already speaks the replacement flow.'];
  }
  const days = Math.floor((((now ?? Date.now() / 1000)) - newest) / 86400);
  return ['silent-since-sunset',
    `newest upload is ${days} day(s) old and predates the cutover. Every caller ` +
    'has been failing since, quietly, at HTTP 200.'];
}

async function get(token, method, params = {}) {
  const url = new URL(API + method);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
  const res = await fetch(url, { headers: { Authorization: `Bearer ${token}` } });
  try {
    return await res.json();
  } catch {
    return null;
  }
}

async function main() {
  const token = process.env.SLACK_BOT_TOKEN;
  if (!token) {
    console.error('set SLACK_BOT_TOKEN (files:read is enough for the corroboration)');
    process.exitCode = 2;
    return;
  }

  const args = process.argv.slice(2);
  const i = args.indexOf('--count');
  const count = i === -1 ? '100' : args[i + 1];

  const [state, detail] = verdict(await get(token, 'files.upload'));
  let bad = 0;
  if (state === 'retired') {
    bad += 1;
    console.warn(`${state.padEnd(19)} ${detail}`);
    console.warn('  repair: files.getUploadURLExternal(filename, length) -> upload the ' +
                 'raw bytes to upload_url -> files.completeUploadExternal(files, channel_id)');
    console.warn('  or use the SDK helper: client.filesUploadV2({...}) / ' +
                 'client.files_upload_v2(...)');
  } else {
    console.warn(`${state.padEnd(19)} ${detail}`);
  }

  const me = await get(token, 'auth.test');
  if (me?.ok === true) {
    const listing = await get(token, 'files.list', { user: me.user_id, count });
    if (listing?.ok === true) {
      const [hstate, hdetail] = uploadActivity(listing.files ?? []);
      if (hstate === 'silent-since-sunset') {
        bad += 1;
        console.warn(`${hstate.padEnd(19)} ${hdetail}`);
      } else {
        console.log(`${hstate.padEnd(19)} ${hdetail}`);
      }
    } else {
      console.log(`${'no-history'.padEnd(19)} files.list did not answer ok: true ` +
                  `(${listing?.error ?? '?'}); the probe above stands on its own`);
    }
  } else {
    console.log(`${'no-history'.padEnd(19)} auth.test did not answer ok: true, so ` +
                'the history check was skipped');
  }

  console.log(`1 method probed, ${bad} finding(s)`);
  process.exitCode = bad ? 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_files_upload_probe import SUNSET, upload_activity, verdict


def test_method_deprecated_is_the_finding():
    state, detail = verdict({"ok": False, "error": "method_deprecated"})
    assert state == "retired"
    assert "2025-11-12" in detail


def test_deprecated_endpoint_is_the_same_finding():
    assert verdict({"ok": False, "error": "deprecated_endpoint"})[0] == "retired"


def test_missing_scope_proves_nothing_about_the_method():
    state, detail = verdict({"ok": False, "error": "missing_scope", "needed": "files:write"})
    assert state == "unknown"
    assert "files:write" in detail


def test_a_parsed_call_means_the_method_still_answers():
    assert verdict({"ok": False, "error": "no_file_data"})[0] == "still-answering"


def test_a_credential_error_is_not_a_deprecation():
    assert verdict({"ok": False, "error": "invalid_auth"})[0] == "auth"


def test_non_json_body_is_not_an_answer():
    assert verdict("<html>proxy</html>")[0] == "unreadable"


def test_history_ending_before_the_cutover_is_a_silent_outage():
    files = [{"created": SUNSET - 86400 * 30}, {"created": SUNSET - 86400 * 400}]
    state, detail = upload_activity(files, now=SUNSET + 86400 * 10)
    assert state == "silent-since-sunset"
    assert "40 day(s)" in detail


def test_an_upload_after_the_cutover_clears_the_history_check():
    files = [{"created": SUNSET - 10}, {"created": SUNSET + 10}]
    assert upload_activity(files, now=SUNSET + 86400)[0] == "uploading"


def test_no_files_is_not_evidence_either_way():
    assert upload_activity([], now=SUNSET)[0] == "no-uploads"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { SUNSET, uploadActivity, verdict } from './slack-files-upload-probe.mjs';

test('method_deprecated is the finding', () => {
  const [state, detail] = verdict({ ok: false, error: 'method_deprecated' });
  assert.equal(state, 'retired');
  assert.match(detail, /2025-11-12/);
});

test('deprecated_endpoint is the same finding', () => {
  assert.equal(verdict({ ok: false, error: 'deprecated_endpoint' })[0], 'retired');
});

test('missing_scope proves nothing about the method', () => {
  const [state, detail] = verdict({ ok: false, error: 'missing_scope', needed: 'files:write' });
  assert.equal(state, 'unknown');
  assert.match(detail, /files:write/);
});

test('a parsed call means the method still answers', () => {
  assert.equal(verdict({ ok: false, error: 'no_file_data' })[0], 'still-answering');
});

test('a credential error is not a deprecation', () => {
  assert.equal(verdict({ ok: false, error: 'invalid_auth' })[0], 'auth');
});

test('non json body is not an answer', () => {
  assert.equal(verdict('<html>proxy</html>')[0], 'unreadable');
});

test('history ending before the cutover is a silent outage', () => {
  const files = [{ created: SUNSET - 86400 * 30 }, { created: SUNSET - 86400 * 400 }];
  const [state, detail] = uploadActivity(files, SUNSET + 86400 * 10);
  assert.equal(state, 'silent-since-sunset');
  assert.match(detail, /40 day\(s\)/);
});

test('an upload after the cutover clears the history check', () => {
  const files = [{ created: SUNSET - 10 }, { created: SUNSET + 10 }];
  assert.equal(uploadActivity(files, SUNSET + 86400)[0], 'uploading');
});

test('no files is not evidence either way', () => {
  assert.equal(uploadActivity([], SUNSET)[0], 'no-uploads');
});