feat(rotators): GitHub fine-grained PAT rotator + thin GitHub API client
Github::Client wraps the bearer-auth REST API with required 'X-GitHub-Api-Version: 2022-11-28' header. Implements me, create_pat (POST /user/personal-access-tokens), delete_pat (DELETE by id). GithubPatRotator's 4-step contract: - generate: create_pat returns new id + token value (fresh PAT alongside old) - apply: no-op (create_pat already active) - verify: probe-client uses the NEW PAT to GET /user; success means it works - commit: delete OLD PAT by id (tracked via old_pat_id tag) - rollback_apply: delete NEW PAT 7 unit specs verify create/delete/me round-trips, full rotation path, verify-on-401, and rollback deletion.
This commit is contained in:
CarterPerez-dev
parent
d431e9014e
commit
873257ad4b
|
|
@ -0,0 +1,51 @@
|
|||
# ===================
|
||||
# ©AngelaMos | 2026
|
||||
# client_spec.cr
|
||||
# ===================
|
||||
|
||||
require "../../spec_helper"
|
||||
require "webmock"
|
||||
require "../../../src/cre/github/client"
|
||||
|
||||
WebMock.allow_net_connect = false
|
||||
|
||||
private def fresh_client
|
||||
CRE::Github::Client.new(token: "ghp_admin")
|
||||
end
|
||||
|
||||
describe CRE::Github::Client do
|
||||
before_each { WebMock.reset }
|
||||
|
||||
it "creates a fine-grained PAT" do
|
||||
WebMock.stub(:post, "https://api.github.com/user/personal-access-tokens")
|
||||
.with(headers: {"Authorization" => "Bearer ghp_admin"})
|
||||
.to_return(body: %({"id":12345,"token":"ghp_newvalue","expires_at":"2026-07-01T00:00:00Z"}))
|
||||
|
||||
token = fresh_client.create_pat("my-pat", ["repo", "read:org"], 90)
|
||||
token.id.should eq 12345_i64
|
||||
token.token_value.should eq "ghp_newvalue"
|
||||
token.expires_at.should eq "2026-07-01T00:00:00Z"
|
||||
end
|
||||
|
||||
it "deletes a PAT" do
|
||||
deleted = false
|
||||
WebMock.stub(:delete, "https://api.github.com/user/personal-access-tokens/12345")
|
||||
.with(headers: {"Authorization" => "Bearer ghp_admin"})
|
||||
.to_return { |_| deleted = true; HTTP::Client::Response.new(200, body: "{}") }
|
||||
fresh_client.delete_pat(12345_i64)
|
||||
deleted.should be_true
|
||||
end
|
||||
|
||||
it "fetches the authenticated user" do
|
||||
WebMock.stub(:get, "https://api.github.com/user")
|
||||
.to_return(body: %({"login":"octocat","id":1}))
|
||||
user = fresh_client.me
|
||||
user["login"].as_s.should eq "octocat"
|
||||
end
|
||||
|
||||
it "raises GithubError on non-2xx" do
|
||||
WebMock.stub(:get, "https://api.github.com/user")
|
||||
.to_return(status: 401, body: %({"message":"Bad credentials"}))
|
||||
expect_raises(CRE::Github::GithubError) { fresh_client.me }
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,87 @@
|
|||
# ===================
|
||||
# ©AngelaMos | 2026
|
||||
# github_pat_spec.cr
|
||||
# ===================
|
||||
|
||||
require "../../spec_helper"
|
||||
require "webmock"
|
||||
require "../../../src/cre/rotators/github_pat"
|
||||
|
||||
WebMock.allow_net_connect = false
|
||||
|
||||
private def github_credential(old_pat_id : String = "111")
|
||||
CRE::Domain::Credential.new(
|
||||
id: UUID.random,
|
||||
external_id: "deploy-bot",
|
||||
kind: CRE::Domain::CredentialKind::GithubPat,
|
||||
name: "deploy-bot",
|
||||
tags: {
|
||||
"name" => "deploy-bot",
|
||||
"scopes" => %(["repo","read:org"]),
|
||||
"old_pat_id" => old_pat_id,
|
||||
"expires_in_days" => "90",
|
||||
} of String => String,
|
||||
)
|
||||
end
|
||||
|
||||
private def gh_client
|
||||
CRE::Github::Client.new(token: "ghp_admin")
|
||||
end
|
||||
|
||||
describe CRE::Rotators::GithubPatRotator do
|
||||
before_each { WebMock.reset }
|
||||
|
||||
it "executes the full 4-step contract" do
|
||||
cred = github_credential
|
||||
|
||||
WebMock.stub(:post, "https://api.github.com/user/personal-access-tokens")
|
||||
.to_return(body: %({"id":99999,"token":"ghp_new","expires_at":"2026-07-01T00:00:00Z"}))
|
||||
|
||||
rotator = CRE::Rotators::GithubPatRotator.new(gh_client)
|
||||
rotator.can_rotate?(cred).should be_true
|
||||
|
||||
new_secret = rotator.generate(cred)
|
||||
new_secret.metadata["new_pat_id"].should eq "99999"
|
||||
new_secret.metadata["old_pat_id"].should eq "111"
|
||||
String.new(new_secret.ciphertext).should eq "ghp_new"
|
||||
|
||||
rotator.apply(cred, new_secret) # no-op
|
||||
|
||||
WebMock.stub(:get, "https://api.github.com/user")
|
||||
.with(headers: {"Authorization" => "Bearer ghp_new"})
|
||||
.to_return(body: %({"login":"deploy-bot"}))
|
||||
rotator.verify(cred, new_secret).should be_true
|
||||
|
||||
deleted_old = false
|
||||
WebMock.stub(:delete, "https://api.github.com/user/personal-access-tokens/111")
|
||||
.to_return { |_| deleted_old = true; HTTP::Client::Response.new(200, body: "{}") }
|
||||
rotator.commit(cred, new_secret)
|
||||
deleted_old.should be_true
|
||||
end
|
||||
|
||||
it "verify returns false when /user fails with new token" do
|
||||
cred = github_credential
|
||||
WebMock.stub(:get, "https://api.github.com/user")
|
||||
.to_return(status: 401, body: %({"message":"bad"}))
|
||||
rotator = CRE::Rotators::GithubPatRotator.new(gh_client)
|
||||
s = CRE::Domain::NewSecret.new(
|
||||
ciphertext: "ghp_bad".to_slice,
|
||||
metadata: {"new_pat_id" => "1", "old_pat_id" => "0"},
|
||||
)
|
||||
rotator.verify(cred, s).should be_false
|
||||
end
|
||||
|
||||
it "rollback_apply deletes the new PAT" do
|
||||
cred = github_credential
|
||||
rotator = CRE::Rotators::GithubPatRotator.new(gh_client)
|
||||
s = CRE::Domain::NewSecret.new(
|
||||
ciphertext: "ghp_new".to_slice,
|
||||
metadata: {"new_pat_id" => "888"},
|
||||
)
|
||||
deleted = false
|
||||
WebMock.stub(:delete, "https://api.github.com/user/personal-access-tokens/888")
|
||||
.to_return { |_| deleted = true; HTTP::Client::Response.new(200, body: "{}") }
|
||||
rotator.rollback_apply(cred, s)
|
||||
deleted.should be_true
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,82 @@
|
|||
# ===================
|
||||
# ©AngelaMos | 2026
|
||||
# client.cr
|
||||
# ===================
|
||||
|
||||
require "http/client"
|
||||
require "json"
|
||||
|
||||
module CRE::Github
|
||||
class GithubError < Exception
|
||||
getter status : Int32
|
||||
|
||||
def initialize(message : String, @status : Int32)
|
||||
super(message)
|
||||
end
|
||||
end
|
||||
|
||||
# Thin GitHub REST client. We target fine-grained PATs for rotation since the
|
||||
# /user/personal-access-tokens endpoint accepts programmatic creation /
|
||||
# deletion when the bearer token has the appropriate Apps-managed permission.
|
||||
# For the test/portfolio path we mock these endpoints directly.
|
||||
class Client
|
||||
record Token, id : Int64, token_value : String, expires_at : String?
|
||||
|
||||
DEFAULT_API = "https://api.github.com"
|
||||
|
||||
def initialize(@token : String, @api_base : String = DEFAULT_API)
|
||||
end
|
||||
|
||||
def me : JSON::Any
|
||||
get("/user")
|
||||
end
|
||||
|
||||
def create_pat(name : String, scopes : Array(String), expires_in_days : Int32 = 90) : Token
|
||||
payload = {
|
||||
"name" => name,
|
||||
"expires_in_days" => expires_in_days,
|
||||
"scopes" => scopes,
|
||||
}.to_json
|
||||
json = post("/user/personal-access-tokens", payload)
|
||||
Token.new(
|
||||
id: json["id"].as_i64,
|
||||
token_value: json["token"].as_s,
|
||||
expires_at: json["expires_at"]?.try(&.as_s),
|
||||
)
|
||||
end
|
||||
|
||||
def delete_pat(token_id : Int64) : Nil
|
||||
delete("/user/personal-access-tokens/#{token_id}")
|
||||
end
|
||||
|
||||
private def get(path : String) : JSON::Any
|
||||
response = HTTP::Client.get(url(path), headers: headers)
|
||||
raise GithubError.new("GET #{path}: #{response.body[0, 200]?}", response.status_code) unless response.status_code < 300
|
||||
JSON.parse(response.body)
|
||||
end
|
||||
|
||||
private def post(path : String, body : String) : JSON::Any
|
||||
response = HTTP::Client.post(url(path), headers: headers, body: body)
|
||||
raise GithubError.new("POST #{path}: #{response.body[0, 200]?}", response.status_code) unless response.status_code < 300
|
||||
JSON.parse(response.body)
|
||||
end
|
||||
|
||||
private def delete(path : String) : Nil
|
||||
response = HTTP::Client.delete(url(path), headers: headers)
|
||||
raise GithubError.new("DELETE #{path}: #{response.body[0, 200]?}", response.status_code) unless response.status_code < 300
|
||||
end
|
||||
|
||||
private def headers : HTTP::Headers
|
||||
HTTP::Headers{
|
||||
"Authorization" => "Bearer #{@token}",
|
||||
"Accept" => "application/vnd.github+json",
|
||||
"X-GitHub-Api-Version" => "2022-11-28",
|
||||
"Content-Type" => "application/json",
|
||||
}
|
||||
end
|
||||
|
||||
private def url(path : String) : String
|
||||
"#{@api_base}#{path}"
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
# ===================
|
||||
# ©AngelaMos | 2026
|
||||
# github_pat.cr
|
||||
# ===================
|
||||
|
||||
require "json"
|
||||
require "../github/client"
|
||||
require "./rotator"
|
||||
|
||||
module CRE::Rotators
|
||||
# GithubPatRotator manages fine-grained Personal Access Tokens. It uses an
|
||||
# admin/issuer bearer to create the new PAT and to delete the old one.
|
||||
#
|
||||
# Required Credential.tags:
|
||||
# "name" - PAT label
|
||||
# "old_pat_id" - the GitHub PAT id to revoke on commit
|
||||
# "scopes" - JSON-encoded array of scope strings (e.g. ["repo","read:org"])
|
||||
# Optional "expires_in_days" - default 90
|
||||
class GithubPatRotator < Rotator
|
||||
register_as :github_pat
|
||||
|
||||
def initialize(@client : Github::Client)
|
||||
end
|
||||
|
||||
def kind : Symbol
|
||||
:github_pat
|
||||
end
|
||||
|
||||
def can_rotate?(c : Domain::Credential) : Bool
|
||||
c.kind.github_pat? && !c.tag("name").nil? && !c.tag("scopes").nil?
|
||||
end
|
||||
|
||||
def generate(c : Domain::Credential) : Domain::NewSecret
|
||||
raise RotatorError.new("missing 'name' or 'scopes' tag") unless can_rotate?(c)
|
||||
scopes = Array(String).from_json(c.tag("scopes").not_nil!)
|
||||
expires_in_days = (c.tag("expires_in_days") || "90").to_i
|
||||
|
||||
new_token = @client.create_pat(c.tag("name").not_nil!, scopes, expires_in_days)
|
||||
|
||||
Domain::NewSecret.new(
|
||||
ciphertext: new_token.token_value.to_slice,
|
||||
metadata: {
|
||||
"new_pat_id" => new_token.id.to_s,
|
||||
"old_pat_id" => c.tag("old_pat_id") || "",
|
||||
"expires_at" => new_token.expires_at || "",
|
||||
},
|
||||
)
|
||||
end
|
||||
|
||||
def apply(c : Domain::Credential, s : Domain::NewSecret) : Nil
|
||||
_ = {c, s}
|
||||
# No-op: create_pat already exposed the token. Old PAT still works.
|
||||
end
|
||||
|
||||
def verify(c : Domain::Credential, s : Domain::NewSecret) : Bool
|
||||
_ = c
|
||||
probe = Github::Client.new(String.new(s.ciphertext))
|
||||
probe.me
|
||||
true
|
||||
rescue
|
||||
false
|
||||
end
|
||||
|
||||
def commit(c : Domain::Credential, s : Domain::NewSecret) : Nil
|
||||
_ = c
|
||||
old = s.metadata["old_pat_id"]?
|
||||
return if old.nil? || old.empty?
|
||||
@client.delete_pat(old.to_i64)
|
||||
end
|
||||
|
||||
def rollback_apply(c : Domain::Credential, s : Domain::NewSecret) : Nil
|
||||
_ = c
|
||||
new_id = s.metadata["new_pat_id"]?
|
||||
return if new_id.nil? || new_id.empty?
|
||||
@client.delete_pat(new_id.to_i64) rescue nil
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Reference in New Issue