diff --git a/github_copilot_seats/README.md b/github_copilot_seats/README.md index bd73be2..6e60fa0 100644 --- a/github_copilot_seats/README.md +++ b/github_copilot_seats/README.md @@ -1,29 +1,33 @@ --- -title: Unregister non active users from Copilot -description: If Copilot was not used in a preceding period by users, the workflow automatically unregisters and notifies them. Users can ask for their subscription to be reinstated. +title: Cancel GitHub Copilot access for inactive users +description: If Copilot was not used in a preceding period by users, unsubscribe and notify them in Slack. Users can ask for their subscription to be reinstated. integrations: ["githubcopilot", "slack"] categories: ["DevOps"] --- -# GitHub Copilot Registration Pruning +# GitHub Copilot Seat Pruning -This automation searches daily for all users in a GitHub organization that are actively using Copilot. -If Copilot was not used in a preceding period, it automatically unregisters them, and then notifies them. -Users can then optionally ask for their subscription to be reinstated. +This automation enumerates once a day all the users in the GitHub organization +that have access to [Copilot](https://github.com/features/copilot). If any of +them haven't used it in a preceding period of time, it automatically marks +their seat for cancellation in the next billing cycle, and notifies them in a +Slack DM. -## Before Deploying This AutoKitteh Project: +Users can then optionally respond and ask for the seat to be reassigned back +to them. -- Set the `GITHUB_ORG` in the project's vars. -- Set the `IDLE_USAGE_THRESHOLD` in the project's vars: - - (e.g., `4320` for 72 hours) - - (e.g., `25` for 25 minutes) -- Set the `LOGINS` in the project's vars (optional). -- Set the `LOG_CHANNEL` in the project's vars to the Slack channel name/ID you want to post to. +## Before Deploying This AutoKitteh Project + +Set/modify these optional project variables: + +- `IDLE_HOURS_THRESHOLD` (default = 72 hours) +- `MANAGED_LOGINS` (default = no one = manage all org users) +- `SLACK_LOG_CHANNEL` (default = `"copilot-log"`, `""` = no logging in Slack) ## Slack Usage -- `/autokitteh prune-idle-copilot-seats` invokes the automation immediately. -- `/autokitteh find-idle-copilot-seats` displays the potentially idle seats. +You may use the Slack application's slash command(s) with one of these text +triggers: -> [!WARNING] -> This example currently works only with a [Personal Access Token](https://docs.autokitteh.com/integrations/github/connection/#personal-access-token-pat), specifically a [classic token](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens#creating-a-personal-access-token-classic). +- `prune-idle-copilot-seats` - invokes the daily automation immediately +- `find-idle-copilot-seats` - displays the potentially idle seats diff --git a/github_copilot_seats/autokitteh.yaml b/github_copilot_seats/autokitteh.yaml index aa23aab..021d66a 100644 --- a/github_copilot_seats/autokitteh.yaml +++ b/github_copilot_seats/autokitteh.yaml @@ -6,15 +6,13 @@ version: v1 project: name: github_copilot_seats vars: - - name: GITHUB_ORG - value: autokitteh - - name: IDLE_USAGE_THRESHOLD - value: 10 # 72 hours - - # If not empty, only manage GitHub Copilot subscriptions to the specified users, separated by commas. - name: LOGINS + - name: IDLE_HOURS_THRESHOLD + value: 72 + - # Optional: manage GitHub Copilot subscriptions only for these users, separated by commas. + name: MANAGED_LOGINS value: "" - - name: LOG_CHANNEL - value: log + - name: SLACK_LOG_CHANNEL + value: copilot-log connections: - name: github_conn integration: github diff --git a/github_copilot_seats/helpers.py b/github_copilot_seats/helpers.py deleted file mode 100644 index c8b0a21..0000000 --- a/github_copilot_seats/helpers.py +++ /dev/null @@ -1,53 +0,0 @@ -from autokitteh.github import github_client -from autokitteh.slack import slack_client - -github = github_client("github_conn") -slack = slack_client("slack_conn") - - -def _email_to_slack_user_id(email: str) -> str: - """Fetch Slack user ID based on email address.""" - resp = slack.users_lookupByEmail(email=email) - return resp["user"]["id"] if resp.get("ok", False) else None - - -def github_username_to_slack_user_id(username: str) -> str: - """Map a GitHub username to a Slack user ID.""" - resp = github.get_user(username) - - if resp.type == "Bot": - return None - - # Match by email address first. - if resp.email: - slack_user_id = _email_to_slack_user_id(resp.email) - if slack_user_id: - return slack_user_id - - # Match by the user's full name if email doesn't work. - gh_full_name = resp.name.lower() if resp.name else None - if not gh_full_name: - return None - - slack_users = _slack_users() - - # Normalize the names for comparison. - for user in slack_users: - real_name = user.profile.real_name.lower() - normalized_name = user.profile.real_name_normalized.lower() - - if gh_full_name in (real_name, normalized_name): - return user.id - - return None - - -def _slack_users(cursor="") -> list: - """Retrieve all Slack users, handling pagination.""" - resp = slack.users_list(cursor, limit=100) - if not resp.get("ok", False): - return [] - - users = resp["members"] - next_cursor = resp.get("response_metadata", {}).get("next_cursor") - return users + _slack_users(next_cursor) if next_cursor else users diff --git a/github_copilot_seats/msg.json b/github_copilot_seats/msg.json index 8648b20..b6947d5 100644 --- a/github_copilot_seats/msg.json +++ b/github_copilot_seats/msg.json @@ -1,44 +1,42 @@ -{ - "blocks": [ - { - "type": "header", - "text": { - "type": "plain_text", - "text": "You have been removed from the Copilot program due to inactivity", - "emoji": true - } - }, - { - "type": "section", - "text": { - "type": "mrkdwn", - "text": "Please select an option:" - } - }, - { - "type": "actions", - "elements": [ - { - "type": "button", - "text": { - "type": "plain_text", - "text": ":repeat: Reinstate", - "emoji": true - }, - "value": "reinstate", - "action_id": "reinstate-action" - }, - { - "type": "button", - "text": { - "type": "plain_text", - "text": ":relieved: OK", - "emoji": true - }, - "value": "ok", - "action_id": "ok-action" - } - ] - } - ] -} +[ + { + "type": "header", + "text": { + "type": "plain_text", + "text": "You have been removed from the Copilot program due to inactivity", + "emoji": true + } + }, + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": "Please select an option:" + } + }, + { + "type": "actions", + "elements": [ + { + "type": "button", + "text": { + "type": "plain_text", + "text": ":repeat: Reinstate", + "emoji": true + }, + "value": "reinstate", + "action_id": "reinstate-action" + }, + { + "type": "button", + "text": { + "type": "plain_text", + "text": ":relieved: OK", + "emoji": true + }, + "value": "ok", + "action_id": "ok-action" + } + ] + } +] diff --git a/github_copilot_seats/seats.py b/github_copilot_seats/seats.py index eaf206a..179fe47 100644 --- a/github_copilot_seats/seats.py +++ b/github_copilot_seats/seats.py @@ -1,117 +1,114 @@ -from datetime import datetime, timezone, timedelta -import json +from datetime import datetime, timedelta, timezone import os -from typing import Any +from pathlib import Path import autokitteh from autokitteh.github import github_client from autokitteh.slack import slack_client -from helpers import github_username_to_slack_user_id +from users import github_username_to_slack_user_id -ORG_NAME = os.getenv("github_conn__target_name") -IDLE_USAGE_THRESHOLD = int(os.getenv("IDLE_USAGE_THRESHOLD")) -LOGINS = os.getenv("LOGINS") -LOG_CHANNEL = os.getenv("LOG_CHANNEL") +IDLE_HOURS_THRESHOLD = int(os.getenv("IDLE_HOURS_THRESHOLD", "72")) +MANAGED_LOGINS = os.getenv("MANAGED_LOGINS") -logins = LOGINS.split(",") if LOGINS else [] github = github_client("github_conn") -slack = slack_client("slack_conn") - -org = github.get_organization(ORG_NAME) +org = github.get_organization(os.getenv("github_conn__target_name")) copilot = org.get_copilot() +slack = slack_client("slack_conn") -def prune_idle_seats() -> list[dict]: - """Prunes idle GitHub Copilot users based on their last activity time.""" - seats = find_idle_seats() - for seat in seats: - autokitteh.start(loc="seats.py:engage_seat", data=seat) - return seats +def find_idle_seats(prune: bool = False) -> list[dict[str, str]]: + """Identifies idle GitHub Copilot users based on their last activity time. -def find_idle_seats() -> list[dict]: - """Identifies idle GitHub Copilot users based on their last activity time.""" - seats = copilot.get_seats() - t = datetime.now(timezone.utc) + If `prune` is set to `True`, it also cancels their seat assignments and + interacts with the users asynchronously to confirm this action. + """ idle_seats = [] - - for seat in seats: - if logins and seat.assignee.login not in logins: - print(f"skipping {seat.assignee.login}") + for seat in copilot.get_seats(): + # If the project is limited to specific org users, ignore the rest. + managed_logins = MANAGED_LOGINS.split(",") if MANAGED_LOGINS else [] + if managed_logins and seat.assignee.login not in managed_logins: + print(f"Skipping unmanaged user: {seat.assignee.login}") continue - delta = t - seat.last_activity_at - is_idle = delta >= timedelta(minutes=IDLE_USAGE_THRESHOLD) + # Was the assigned seat being used recently? + now = datetime.now(timezone.utc) + delta = now - seat.last_activity_at + is_active = delta < timedelta(hours=IDLE_HOURS_THRESHOLD) - comparison = ">=" if is_idle else "<" - status = f"{seat.assignee.login}: {t} - {seat.last_activity_at} = {delta} {comparison} {IDLE_USAGE_THRESHOLD} minutes" + comparison = "<" if is_active else ">=" + status = f"{seat.assignee.login}: {now} - {seat.last_activity_at} = " + status += f"{delta} {comparison} {IDLE_HOURS_THRESHOLD} hours" print(status) - if is_idle: - # Convert CopilotSeat object to a dictionary - seat_dict = { - "assignee": {"login": seat.assignee.login}, - "last_activity_at": seat.last_activity_at.isoformat(), - } - idle_seats.append(seat_dict) + if is_active: + continue + + # Convert the non-serializable "CopilotSeat" object into a simple dictionary. + simple_seat = { + "assignee_login": seat.assignee.login, + "last_activity_at": seat.last_activity_at.isoformat(), + } + idle_seats.append(simple_seat) + + # Interact with the user asynchronously in a child workflow. + if prune: + autokitteh.start(loc="seats.py:prune_idle_seat", data=simple_seat) return idle_seats -def engage_seat(seat: dict[str, Any]) -> None: - """Engages a GitHub user assigned to a seat by identifying their corresponding Slack user and initiating a workflow. +def prune_idle_seat(seat: dict[str, str]) -> None: + """Interacts via Slack with a GitHub user assigned to an idle Copilot seat. - Note: - This is designed to run as a child workflow using: - autokitteh.start(loc="seats.py:engage_seat", data={"seat": seat}) + Note - this function runs as a child workflow, by calling: + autokitteh.start(loc="filename.py:function_name", data={...}) Args: - seat (dict): Contains details about the assigned GitHub user. + seat: Username and last activity timestamp of the GitHub user to which + the Copilot seat is assigned. """ - github_login = seat["assignee"]["login"] - # FOR TESTING ONLY; REMOVE AFTER TESTING - if github_login != "pashafateev": - return - report(github_login, "engaging") + github_login = seat["assignee_login"] + report(github_login, "removing seat") + copilot.remove_seats([github_login]) slack_id = github_username_to_slack_user_id(github_login) if not slack_id: - print(f"No slack user found for GitHub user {github_login}") + print(f"No Slack user found for GitHub user {github_login}") return - copilot.remove_seats([github_login]) + report(github_login, "notifying user") + + # Load a blocks-based interactive message template + # from a JSON file and post it to the user's Slack. + slack.chat_postMessage(channel=slack_id, blocks=Path("msg.json").read_text()) + + # Subscribe to Slack interaction events, waiting for the user's response. + filter = f"event_type = 'interaction' && data.user.id == '{slack_id}'" + subscription = autokitteh.subscribe("slack_conn", filter) + + # Retrieve the value from the user's response in the Slack event. + value = autokitteh.next_event(subscription)["actions"][0]["value"] + + # The user's response either confirms the action or reinstates the seat. + match value: + case "ok": + report(github_login, "ok") + msg = "Okey dokey!" + case "reinstate": + report(github_login, "reinstate") + copilot.add_seats([github_login]) + msg = "You have been reinstated to the Copilot program" + case _: + report(github_login, f"weird response: {value}") + msg = f"Response: `{value}` not recognized." - # Loads a predefined message (blocks) from a JSON file and posts it to the user's Slack - with open("msg.json") as file: - msg = json.load(file) - slack.chat_postMessage(channel=slack_id, blocks=msg["blocks"]) - - # Subscribes to Slack interaction events, waiting for the user's response - s = autokitteh.subscribe( - "slack_conn", f'data.type == "block_actions" && data.user.id == "{slack_id}"' - ) - - # Retrieves the value from the user's response in the Slack event - value = autokitteh.next_event(s)["actions"][0]["value"] - - # Based on the user's response, it either confirms the action or reinstates the seat - if value == "ok": - slack.chat_postMessage(channel=slack_id, text="Okey dokey!") - report(github_login, "ok") - elif value == "reinstate": - report(github_login, "reinstate") - copilot.add_seats([github_login]) - slack.chat_postMessage( - channel=slack_id, text="You have been reinstated to the Copilot program." - ) - else: - report(github_login, f"weird response: {value}") - slack.chat_postMessage( - channel=slack_id, text=f"Response: {value} not recognized." - ) + slack.chat_postMessage(channel=slack_id, text=msg) def report(github_login: str, msg: str) -> None: - slack.chat_postMessage(channel=LOG_CHANNEL, text=f"{github_login}: {msg}") + channel = os.getenv("SLACK_LOG_CHANNEL") + if channel: + slack.chat_postMessage(channel=channel, text=f"{github_login}: {msg}") diff --git a/github_copilot_seats/triggers.py b/github_copilot_seats/triggers.py index f7cedc7..9ec0305 100644 --- a/github_copilot_seats/triggers.py +++ b/github_copilot_seats/triggers.py @@ -2,29 +2,34 @@ import seats -s = slack_client("slack_conn") - def on_schedule() -> None: - print(seats.prune_idle_seats()) + for seat in seats.find_idle_seats(prune=True): + print(seat) def on_slack_slash_command(event) -> None: - cmd = event.data.text.lower() - cid = event.data.channel_id - uid = event.data.user_id - - if cmd == "prune-idle-copilot-seats": - idle_seats = seats.prune_idle_seats() - msg = f"Engaged {len(idle_seats)} new idle seats: {', '.join(get_logins(idle_seats))}" - s.chat_postEphemeral(channel=cid, user=uid, text=msg) - elif cmd == "find-idle-copilot-seats": - idle_seats = seats.find_idle_seats() - msg = f"Found {len(idle_seats)} idle seats: {', '.join(get_logins(idle_seats))}" - s.chat_postEphemeral(channel=cid, user=uid, text=msg) + find_seats = True + match event.data.text.lower(): + case "prune-idle-copilot-seats": + prune = True + case "find-idle-copilot-seats": + prune = False + case _: + find_seats = False + + if find_seats: + idle_seats = seats.find_idle_seats(prune) + action = "Pruned" if prune else "Found" + msg = f"{action} {len(idle_seats)} idle seats for these users: " + msg += ", ".join(_get_logins(idle_seats)) else: - s.chat_postEphemeral(channel=cid, user=uid, text="Unrecognized command") + msg = "Error: unrecognized command" + + slack_client("slack_conn").chat_postEphemeral( + channel=event.data.channel_id, user=event.data.user_id, text=msg + ) -def get_logins(seat_list: list) -> list: - return [seat["assignee"]["login"] for seat in seat_list] +def _get_logins(idle_seats: list[dict[str, str]]) -> list[str]: + return [seat["assignee_login"] for seat in idle_seats] diff --git a/github_copilot_seats/users.py b/github_copilot_seats/users.py new file mode 100644 index 0000000..fc7f1b2 --- /dev/null +++ b/github_copilot_seats/users.py @@ -0,0 +1,69 @@ +"""User-related helper functions across GitHub and Slack. + +Based on: https://github.com/autokitteh/kittehub/blob/main/purrr/users.py +""" + +from autokitteh.github import github_client +from autokitteh.slack import slack_client +from slack_sdk.errors import SlackApiError + + +github = github_client("github_conn") +slack = slack_client("slack_conn") + + +def github_username_to_slack_user_id(github_username: str) -> str: + """Convert a GitHub username to a Slack user ID, or "" if not found. + + This function tries to match the email address first, and then + falls back to matching the user's full name (case-insensitive). + """ + user = github.get_user(github_username) + + # Special case: GitHub bots can't have Slack identities. + if user.type == "Bot": + return "" + + # Try to match by the email address first. + if user.email: + slack_user_id = _email_to_slack_user_id(user.email) + if slack_user_id: + return slack_user_id + + # Otherwise, try to match by the user's full name. + github_name = (user.name or "").lower() + if not github_name: + return "" + + for user in _slack_users(): + profile = user.get("profile", {}) + real_name = profile.get("real_name", "").lower() + normalized_name = profile.get("real_name_normalized", "").lower() + if github_name in (real_name, normalized_name): + return user.id + + return "" + + +def _email_to_slack_user_id(email: str) -> str: + """Convert an email address to a Slack user ID, or "" if not found.""" + try: + resp = slack.users_lookupByEmail(email=email) + return resp.get("user", {}).get("id", "") + except SlackApiError: + return "" + + +def _slack_users() -> list[dict]: + """Return a list of all Slack users in the workspace.""" + users = [] + next_cursor = None + while next_cursor != "": + try: + resp = slack.users_list(cursor=next_cursor, limit=100) + users += resp.get("members", []) + next_cursor = resp.get("response_metadata", {}).get("next_cursor", "") + except SlackApiError: + next_cursor = "" + + return users diff --git a/purrr/users.py b/purrr/users.py index c14f6f5..000dab2 100644 --- a/purrr/users.py +++ b/purrr/users.py @@ -4,8 +4,8 @@ import github from slack_sdk.errors import SlackApiError -import debug import data_helper +import debug import github_helper @@ -162,19 +162,13 @@ def github_username_to_slack_user(github_username: str) -> dict: def github_username_to_slack_user_id(github_username: str) -> str: - """Convert a GitHub username to a Slack user ID. + """Convert a GitHub username to a Slack user ID, or "" if not found. This function tries to match the email address first, and then falls back to matching the user's full name (case-insensitive). This function also caches both successful and failed results for a day, to reduce the amount of API calls, especially to Slack. - - Args: - github_username: GitHub username. - - Returns: - Slack user ID, or "" if not found. """ # Don't even check GitHub teams, only individual users. if "/" in github_username: @@ -212,11 +206,9 @@ def github_username_to_slack_user_id(github_username: str) -> str: for user in _slack_users(): profile = user.get("profile", {}) - slack_names = ( - profile.get("real_name", "").lower(), - profile.get("real_name_normalized", "").lower(), - ) - if github_name in slack_names: + real_name = profile.get("real_name", "").lower() + normalized_name = profile.get("real_name_normalized", "").lower() + if github_name in (real_name, normalized_name): data_helper.cache_slack_user_id(github_username, user.id) return user.id