Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/dependency-audit.yml
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
name: Dependency Audit

on:
workflow_dispatch:
pull_request:
paths:
- "requirements.txt"
Expand Down
3 changes: 2 additions & 1 deletion app.py
Original file line number Diff line number Diff line change
Expand Up @@ -921,13 +921,14 @@ def configure_pat():

# Immediately mint a controlled short-lived token from the user-pasted PAT.
# This gives us a token ID we own — all future rotations can revoke the old one.
# The user-pasted PAT becomes unused after this (expires per its own lifetime).
os.environ["DATABRICKS_TOKEN"] = token
pat_rotator._current_token = token
pat_rotator._current_token_id = None
rotated = pat_rotator._rotate_once()
if rotated:
token = pat_rotator.token # use the newly minted token from here on
# Revoke only the bootstrap PAT — leave other user PATs intact (#98)
pat_rotator.revoke_bootstrap_token()
else:
# Rotation failed — fall back to user-pasted token (still valid)
pat_rotator._write_databrickscfg(token)
Expand Down
58 changes: 58 additions & 0 deletions pat_rotator.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,64 @@ def _rotate_once(self):

return True

def revoke_bootstrap_token(self):
"""Revoke only the bootstrap PAT after the first rotation.

Called once after the bootstrap PAT is replaced by a controlled
short-lived token. Lists all tokens, identifies the bootstrap
as the most-recently-created token without a "coda-auto-rotated"
comment, and revokes only that one. Other user PATs (notebooks,
CI, etc.) are left untouched.
"""
current_id = self._current_token_id
token = self._current_token
if not token or not current_id:
return

try:
resp = requests.get(
f"{self._host}/api/2.0/token/list",
headers={"Authorization": f"Bearer {token}"},
timeout=30
)
if resp.status_code != 200:
logger.warning(f"Bootstrap cleanup: failed to list tokens ({resp.status_code})")
return
except requests.RequestException as e:
logger.warning(f"Bootstrap cleanup: list request failed: {e}")
return

token_infos = resp.json().get("token_infos", [])

# Find the bootstrap PAT: newest non-coda token that isn't the current one
candidates = [
info for info in token_infos
if info.get("token_id") != current_id
and info.get("comment", "") != "coda-auto-rotated"
]
if not candidates:
logger.info("Bootstrap cleanup: no bootstrap token candidate found")
return

# The bootstrap PAT is the most recently created candidate
bootstrap = max(candidates, key=lambda t: t.get("creation_time", 0))
tid = bootstrap.get("token_id")
comment = bootstrap.get("comment", "(no comment)")

try:
del_resp = requests.post(
f"{self._host}/api/2.0/token/delete",
headers={"Authorization": f"Bearer {token}"},
json={"token_id": tid},
timeout=30
)
if del_resp.status_code == 200:
logger.info(f"Bootstrap cleanup: revoked bootstrap PAT {tid} ({comment})")
else:
logger.warning(f"Bootstrap cleanup: failed to revoke {tid} ({del_resp.status_code})")
except requests.RequestException as e:
logger.warning(f"Bootstrap cleanup: revoke request failed: {e}")

def _persist_token(self, token):
"""Write rotated token to all persistence layers."""
os.environ["DATABRICKS_TOKEN"] = token
Expand Down