a link shortening certificate expires and short links break

"""Watch the certificate on a Twilio link-shortening domain for expiry.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. Uploading a certificate is a change to how
your customers' links terminate TLS, so the replacement is printed and a person
performs it.
"""
import argparse
import datetime
import logging
import os
import sys

import requests

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

MSG = "https://messaging.twilio.com/v1"
MONITOR = "https://monitor.twilio.com/v1"

# 30131 is the early warning and is logged at warning level; 30120 and 30129 are
# the hard failures. Sweeping only LogLevel=error throws the lead time away.
LINK_ERRORS = (30120, 30129, 30131)


def parse_time(value):
    """Parse a messaging v1 timestamp. Pure.

    These come back as ISO 8601 with a trailing Z, which
    datetime.fromisoformat did not accept before Python 3.11.
    """
    text = str(value or "").strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        return datetime.datetime.fromisoformat(text)
    except ValueError:
        return None


def days_left(date_expires, now):
    """Days until the certificate expires. Negative once it has passed."""
    expires = parse_time(date_expires)
    if expires is None or now is None:
        return None
    return (expires - now).total_seconds() / 86400.0


def validation_pending(cert):
    """True when a replacement has been uploaded and is not validated yet. Pure."""
    pending = (cert or {}).get("cert_in_validation")
    if not pending:
        return False
    return str(pending.get("status") or "").lower() != "validated"


def verdict(cert, days, window_days=30):
    """Classify one link-shortening domain certificate. Pure.

    `days` is the time remaining, or None. Taking it as an argument keeps the
    clock out of the classifier. Returns (state, detail).
    """
    if not cert:
        return ("no-certificate",
                "no certificate on this domain. That is what a Twilio-managed "
                "domain looks like from here, and also what a wrong domain sid "
                "looks like. Confirm which before treating it as clean.")

    pending = validation_pending(cert)

    if days is None:
        return ("expiry-unreadable",
                "a certificate is present and date_expires could not be read, "
                "so nothing can be said about when it lapses.")

    if days <= 0:
        return ("expired",
                "date_expires passed %.0f days ago. Shortened links are "
                "failing TLS in the browser and sends are returning 30120 or "
                "30129." % abs(days))

    if days <= window_days and pending:
        return ("expiring-replacement-validating",
                "%.0f days left, and cert_in_validation is not validated. A "
                "replacement has been uploaded but it is not live yet, so the "
                "clock is still running on the old one." % days)

    if days <= window_days:
        return ("expiring",
                "%.0f days left, inside the %d day renewal window. 30131 will "
                "appear first, at warning level." % (days, window_days))

    if pending:
        return ("validation-pending",
                "the live certificate has %.0f days left, but a replacement in "
                "cert_in_validation is not validated. Worth finishing rather "
                "than leaving half done." % days)

    return ("current", "%.0f days left on the certificate." % days)


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    if r.status_code == 404:
        return None
    r.raise_for_status()
    return r.json()


def link_alerts(session, since, levels=("error", "warning"), limit=2000):
    """Alerts carrying a link-shortening error code, at both log levels.

    Alerts are retained 30 days, which bounds what this can corroborate.
    """
    found = []
    for level in levels:
        page = get(session, MONITOR + "/Alerts", LogLevel=level,
                   StartDate=since, PageSize=100)
        while page:
            for alert in page.get("alerts", []):
                try:
                    code = int(alert.get("error_code") or 0)
                except (TypeError, ValueError):
                    continue
                if code in LINK_ERRORS:
                    found.append(alert)
            nxt = (page.get("meta") or {}).get("next_page_url")
            if not nxt or len(found) >= limit:
                break
            page = get(session, nxt)
    return found


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--domain-sid", action="append", default=[],
                    help="link-shortening domain sid; repeatable. From Console, "
                         "Messaging, Link Shortening")
    ap.add_argument("--window-days", type=int, default=30,
                    help="renewal window: long enough to find the key holder, "
                         "reissue, upload and validate")
    ap.add_argument("--alert-days", type=int, default=7)
    args = ap.parse_args()

    if not args.domain_sid:
        log.error("pass at least one --domain-sid; there is no account-wide "
                  "list of link-shortening domains read here")
        return 2

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)
    now = datetime.datetime.now(datetime.timezone.utc)
    since = (now - datetime.timedelta(days=args.alert_days)).date().isoformat()

    alerts = link_alerts(session, since)
    if alerts:
        codes = sorted({str(a.get("error_code")) for a in alerts})
        log.warning("%d link-shortening alert(s) in the last %d days, codes %s",
                    len(alerts), args.alert_days, ", ".join(codes))

    bad = 0
    for sid in args.domain_sid:
        cert = get(session, "%s/LinkShortening/Domains/%s/Certificate" % (MSG, sid))
        days = days_left((cert or {}).get("date_expires"), now)
        state, detail = verdict(cert, days, args.window_days)
        line = "%-32s %s  %s" % (state, sid, detail)
        if state == "current":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        if state in ("expired", "expiring", "expiring-replacement-validating",
                     "expiry-unreadable"):
            log.warning("  repair: upload a fresh TlsCert to %s/LinkShortening/"
                        "Domains/%s/Certificate, or move the domain to "
                        "Twilio-managed certificates in Console, Messaging, "
                        "Link Shortening, which removes this clock entirely",
                        MSG, sid)
        elif state == "validation-pending":
            log.warning("  repair: finish validating the replacement on %s "
                        "rather than leaving two certificates half swapped", sid)

    log.info("%d domain(s), %d needing a certificate", len(args.domain_sid), bad)
    return 1 if bad else 0


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

/**
 * Watch the certificate on a Twilio link-shortening domain for expiry.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The replacement is printed.
 */
const MSG = 'https://messaging.twilio.com/v1';
const MONITOR = 'https://monitor.twilio.com/v1';

// 30131 is the early warning and is logged at warning level; 30120 and 30129
// are the hard failures. Sweeping only LogLevel=error throws the lead time away.
export const LINK_ERRORS = [30120, 30129, 30131];

/** Parse a messaging v1 ISO 8601 timestamp. Pure. Returns a Date or null. */
export function parseTime(value) {
  const text = String(value ?? '').trim();
  if (!text) return null;
  const t = Date.parse(text);
  return Number.isNaN(t) ? null : new Date(t);
}

/** Days until the certificate expires. Negative once it has passed. */
export function daysLeft(dateExpires, now) {
  const expires = parseTime(dateExpires);
  if (expires === null || !now) return null;
  return (expires.getTime() - now.getTime()) / 86400000;
}

/** True when a replacement has been uploaded and is not validated yet. Pure. */
export function validationPending(cert) {
  const pending = (cert ?? {}).cert_in_validation;
  if (!pending) return false;
  return String(pending.status ?? '').toLowerCase() !== 'validated';
}

/**
 * Classify one link-shortening domain certificate. `days` is the time
 * remaining, or null; taking it as an argument keeps the clock out of the
 * classifier. Pure. Returns [state, detail].
 */
export function verdict(cert, days, windowDays = 30) {
  if (!cert) {
    return ['no-certificate',
      'no certificate on this domain. That is what a Twilio-managed domain ' +
      'looks like from here, and also what a wrong domain sid looks like. ' +
      'Confirm which before treating it as clean.'];
  }

  const pending = validationPending(cert);

  if (days === null || days === undefined) {
    return ['expiry-unreadable',
      'a certificate is present and date_expires could not be read, so nothing ' +
      'can be said about when it lapses.'];
  }

  if (days <= 0) {
    return ['expired',
      `date_expires passed ${Math.abs(days).toFixed(0)} days ago. Shortened ` +
      'links are failing TLS in the browser and sends are returning 30120 or 30129.'];
  }

  if (days <= windowDays && pending) {
    return ['expiring-replacement-validating',
      `${days.toFixed(0)} days left, and cert_in_validation is not validated. A ` +
      'replacement has been uploaded but it is not live yet, so the clock is ' +
      'still running on the old one.'];
  }

  if (days <= windowDays) {
    return ['expiring',
      `${days.toFixed(0)} days left, inside the ${windowDays} day renewal ` +
      'window. 30131 will appear first, at warning level.'];
  }

  if (pending) {
    return ['validation-pending',
      `the live certificate has ${days.toFixed(0)} days left, but a replacement ` +
      'in cert_in_validation is not validated. Worth finishing rather than ' +
      'leaving half done.'];
  }

  return ['current', `${days.toFixed(0)} days left on the certificate.`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (res.status === 404) return null;
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

async function linkAlerts(auth, since, levels = ['error', 'warning'], limit = 2000) {
  const found = [];
  for (const level of levels) {
    let page = await get(auth, `${MONITOR}/Alerts`,
                         { LogLevel: level, StartDate: since, PageSize: 100 });
    while (page) {
      for (const alert of page.alerts ?? []) {
        const code = Number.parseInt(alert.error_code ?? 0, 10);
        if (LINK_ERRORS.includes(code)) found.push(alert);
      }
      const nxt = page.meta?.next_page_url;
      if (!nxt || found.length >= limit) break;
      page = await get(auth, nxt);
    }
  }
  return found;
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;

  const sids = [];
  for (let i = 0; i < process.argv.length - 1; i += 1) {
    if (process.argv[i] === '--domain-sid') sids.push(process.argv[i + 1]);
  }
  if (sids.length === 0) {
    console.error('pass at least one --domain-sid; there is no account-wide list ' +
                  'of link-shortening domains read here');
    process.exitCode = 2;
    return;
  }

  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const flag = process.argv.indexOf('--window-days');
  const windowDays = flag >= 0 ? Number(process.argv[flag + 1]) : 30;
  const now = new Date();
  const since = new Date(now.getTime() - 7 * 86400000).toISOString().slice(0, 10);

  const alerts = await linkAlerts(auth, since);
  if (alerts.length) {
    const codes = [...new Set(alerts.map((a) => String(a.error_code)))].sort();
    console.warn(`${alerts.length} link-shortening alert(s) in the last 7 days, ` +
                 `codes ${codes.join(', ')}`);
  }

  let bad = 0;
  for (const sid of sids) {
    const cert = await get(auth, `${MSG}/LinkShortening/Domains/${sid}/Certificate`);
    const days = daysLeft((cert ?? {}).date_expires, now);
    const [state, detail] = verdict(cert, days, windowDays);
    const line = `${state.padEnd(32)} ${sid}  ${detail}`;
    if (state === 'current') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    if (state === 'expired' || state === 'expiring' ||
        state === 'expiring-replacement-validating' || state === 'expiry-unreadable') {
      console.warn(`  repair: upload a fresh TlsCert to ${MSG}/LinkShortening/` +
                   `Domains/${sid}/Certificate, or move the domain to ` +
                   'Twilio-managed certificates in Console, Messaging, Link ' +
                   'Shortening, which removes this clock entirely');
    } else if (state === 'validation-pending') {
      console.warn(`  repair: finish validating the replacement on ${sid} rather ` +
                   'than leaving two certificates half swapped');
    }
  }

  console.log(`${sids.length} domain(s), ${bad} needing a certificate`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing credentials and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

import datetime

from twilio_link_domain_cert_audit import days_left, validation_pending, verdict

CERT = {"date_expires": "2026-10-01T00:00:00Z"}
VALIDATING = dict(CERT, cert_in_validation={"status": "pending"})
VALIDATED = dict(CERT, cert_in_validation={"status": "validated"})
NOW = datetime.datetime(2026, 8, 30, tzinfo=datetime.timezone.utc)


def test_inside_the_renewal_window_is_the_finding():
    state, detail = verdict(CERT, 12.0, window_days=30)
    assert state == "expiring"
    assert "30131" in detail


def test_outside_the_window_is_current():
    assert verdict(CERT, 120.0, window_days=30)[0] == "current"


def test_an_expired_certificate_names_both_failure_codes():
    state, detail = verdict(CERT, -3.0)
    assert state == "expired"
    assert "30120" in detail and "30129" in detail


def test_a_replacement_in_validation_does_not_stop_the_clock():
    # This is the state that reads as handled in a status meeting and is not.
    state, detail = verdict(VALIDATING, 4.0, window_days=30)
    assert state == "expiring-replacement-validating"
    assert "not live yet" in detail


def test_a_stalled_replacement_on_a_healthy_certificate_is_untidy_not_urgent():
    assert verdict(VALIDATING, 200.0)[0] == "validation-pending"


def test_a_validated_replacement_is_not_reported():
    assert verdict(VALIDATED, 200.0)[0] == "current"
    assert validation_pending(VALIDATED) is False


def test_no_certificate_is_reported_as_unknown_rather_than_clean():
    state, detail = verdict(None, None)
    assert state == "no-certificate"
    assert "wrong domain sid" in detail


def test_days_left_reads_the_trailing_z_timestamp():
    assert round(days_left("2026-09-06T00:00:00Z", NOW)) == 7
    assert round(days_left("2026-08-23T00:00:00Z", NOW)) == -7
    assert days_left("not a date", NOW) is None

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { daysLeft, validationPending, verdict } from './twilio-link-domain-cert-audit.mjs';

const CERT = { date_expires: '2026-10-01T00:00:00Z' };
const VALIDATING = { ...CERT, cert_in_validation: { status: 'pending' } };
const VALIDATED = { ...CERT, cert_in_validation: { status: 'validated' } };
const NOW = new Date('2026-08-30T00:00:00Z');

test('inside the renewal window is the finding', () => {
  const [state, detail] = verdict(CERT, 12.0, 30);
  assert.equal(state, 'expiring');
  assert.match(detail, /30131/);
});

test('outside the window is current', () => {
  assert.equal(verdict(CERT, 120.0, 30)[0], 'current');
});

test('an expired certificate names both failure codes', () => {
  const [state, detail] = verdict(CERT, -3.0);
  assert.equal(state, 'expired');
  assert.match(detail, /30120/);
  assert.match(detail, /30129/);
});

test('a replacement in validation does not stop the clock', () => {
  const [state, detail] = verdict(VALIDATING, 4.0, 30);
  assert.equal(state, 'expiring-replacement-validating');
  assert.match(detail, /not live yet/);
});

test('a stalled replacement on a healthy certificate is untidy, not urgent', () => {
  assert.equal(verdict(VALIDATING, 200.0)[0], 'validation-pending');
});

test('a validated replacement is not reported', () => {
  assert.equal(verdict(VALIDATED, 200.0)[0], 'current');
  assert.equal(validationPending(VALIDATED), false);
});

test('no certificate is reported as unknown rather than clean', () => {
  const [state, detail] = verdict(null, null);
  assert.equal(state, 'no-certificate');
  assert.match(detail, /wrong domain sid/);
});

test('daysLeft reads the trailing z timestamp', () => {
  assert.equal(Math.round(daysLeft('2026-09-06T00:00:00Z', NOW)), 7);
  assert.equal(Math.round(daysLeft('2026-08-23T00:00:00Z', NOW)), -7);
  assert.equal(daysLeft('not a date', NOW), null);
});