24067770611

Committed 07 Apr 2026 06:30AM UTC coverage: 91.372% (-0.01%) from 91.383%

Build # 24067770611

Build Type

push

github

Committed by

cuu508

Commit Message

Fix bytes/str mismatch

Coverage Stats

0 of 2 new or added lines in 1 file covered. (0.0%)

8282 of 9064 relevant lines covered (91.37%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.94

/hc/integrations/github/client.py

import logging
import time

import jwt
from django.conf import settings
from pydantic import BaseModel

from hc.lib import curl

logger = logging.getLogger(__name__)


class BadCredentials(Exception):
    pass


class OAuthResponse(BaseModel):
    access_token: str


def get_user_access_token(code: str) -> str:
    """Exchange OAuth code for user access token."""

    # Reference:
    # https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-user-access-token-for-a-github-app

    url = "https://github.com/login/oauth/access_token"
    data = {
        "client_id": settings.GITHUB_CLIENT_ID,
        "client_secret": settings.GITHUB_CLIENT_SECRET,
        "code": code,
    }
    headers = {"Accept": "application/vnd.github+json"}
    result = curl.post(url, data, headers=headers)
    doc = OAuthResponse.model_validate_json(result.content, strict=True)
    return doc.access_token


class Installation(BaseModel):
    id: int


class InstallationsResponse(BaseModel):
    # Error responses contain a "message" field
    message: str | None = None
    installations: list[Installation] | None = None


def get_installation_ids(user_access_token: str) -> list[int]:
    """Retrieve the installation ids the user has access to."""

    # Reference:
    # https://docs.github.com/en/rest/apps/installations?apiVersion=2022-11-28#list-app-installations-accessible-to-the-user-access-token

    url = "https://api.github.com/user/installations"
    headers = {"Authorization": f"Bearer {user_access_token}"}
    result = curl.get(url, headers=headers)
    doc = InstallationsResponse.model_validate_json(result.content, strict=True)
    if doc.message == "Bad credentials":
        # GitHub returns "Bad Credential" response when:
        # - We have acquired a valid user access token,
        # - The user then revokes the access token
        #   (removes us from Settings / Applications / Authorized GitHub Apps)
        # - We then try to use the token to load user's installations
        raise BadCredentials()

    if doc.installations is None:
        content_str = result.content.decode()
        logger.warning(f"Unexpected response from GitHub: {content_str}")

    assert doc.installations is not None
    return [item.id for item in doc.installations]


class Repository(BaseModel):
    full_name: str


class RepositoriesResponse(BaseModel):
    repositories: list[Repository]


def get_repos(user_access_token: str) -> dict[str, int]:
    """Retrieve the repositories the user has access to.

    Return a dict with repo names as keys and the corresponding installation ids
    as values:

        {"owner/repo_name": inst_id, ...}

    """

    # Reference:
    # https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#list-repositories-for-a-user

    results = {}
    for inst_id in get_installation_ids(user_access_token):
        url = f"https://api.github.com/user/installations/{inst_id}/repositories"
        headers = {"Authorization": f"Bearer {user_access_token}"}
        result = curl.get(url, headers=headers)
        doc = RepositoriesResponse.model_validate_json(result.content, strict=True)
        for repo in doc.repositories:
            results[repo.full_name] = inst_id

    return results


class AccessTokensResponse(BaseModel):
    token: str


def get_installation_access_token(installation_id: int) -> str | None:
    """Acquire the installation access token for a specific installation id."""

    # Reference:
    # https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-an-installation-access-token-for-a-github-app

    iat = int(time.time())
    payload = {
        "iat": int(time.time()),
        "exp": iat + 600,
        "iss": settings.GITHUB_CLIENT_ID,
    }

    assert settings.GITHUB_PRIVATE_KEY
    encoded = jwt.encode(payload, settings.GITHUB_PRIVATE_KEY, algorithm="RS256")
    url = f"https://api.github.com/app/installations/{installation_id}/access_tokens"
    result = curl.post(url, headers={"Authorization": f"Bearer {encoded}"})
    if result.status_code == 404:
        # The installation does not exist (our GitHub app has been uninstalled)
        return None

    doc = AccessTokensResponse.model_validate_json(result.content, strict=True)
    return doc.token

1	import logging	1✔
2	import time	1✔
3
4	import jwt	1✔
5	from django.conf import settings	1✔
6	from pydantic import BaseModel	1✔
7
8	from hc.lib import curl	1✔
9
10	logger = logging.getLogger(__name__)	1✔
11
12
13	class BadCredentials(Exception):	1✔
14	pass	1✔
15
16
17	class OAuthResponse(BaseModel):	1✔
18	access_token: str	1✔
19
20
21	def get_user_access_token(code: str) -> str:	1✔
22	"""Exchange OAuth code for user access token."""
23
24	# Reference:
25	# https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-a-user-access-token-for-a-github-app
26
27	url = "https://github.com/login/oauth/access_token"	×
28	data = {	×
29	"client_id": settings.GITHUB_CLIENT_ID,
30	"client_secret": settings.GITHUB_CLIENT_SECRET,
31	"code": code,
32	}
33	headers = {"Accept": "application/vnd.github+json"}	×
34	result = curl.post(url, data, headers=headers)	×
35	doc = OAuthResponse.model_validate_json(result.content, strict=True)	×
36	return doc.access_token	×
37
38
39	class Installation(BaseModel):	1✔
40	id: int	1✔
41
42
43	class InstallationsResponse(BaseModel):	1✔
44	# Error responses contain a "message" field
45	message: str \| None = None	1✔
46	installations: list[Installation] \| None = None	1✔
47
48
49	def get_installation_ids(user_access_token: str) -> list[int]:	1✔
50	"""Retrieve the installation ids the user has access to."""
51
52	# Reference:
53	# https://docs.github.com/en/rest/apps/installations?apiVersion=2022-11-28#list-app-installations-accessible-to-the-user-access-token
54
55	url = "https://api.github.com/user/installations"	×
56	headers = {"Authorization": f"Bearer {user_access_token}"}	×
57	result = curl.get(url, headers=headers)	×
58	doc = InstallationsResponse.model_validate_json(result.content, strict=True)	×
59	if doc.message == "Bad credentials":	×
60	# GitHub returns "Bad Credential" response when:
61	# - We have acquired a valid user access token,
62	# - The user then revokes the access token
63	# (removes us from Settings / Applications / Authorized GitHub Apps)
64	# - We then try to use the token to load user's installations
65	raise BadCredentials()	×
66
67	if doc.installations is None:	×
NEW 68	content_str = result.content.decode()	×
NEW 69	logger.warning(f"Unexpected response from GitHub: {content_str}")	×
70
71	assert doc.installations is not None	×
72	return [item.id for item in doc.installations]	×
73
74
75	class Repository(BaseModel):	1✔
76	full_name: str	1✔
77
78
79	class RepositoriesResponse(BaseModel):	1✔
80	repositories: list[Repository]	1✔
81
82
83	def get_repos(user_access_token: str) -> dict[str, int]:	1✔
84	"""Retrieve the repositories the user has access to.
85
86	Return a dict with repo names as keys and the corresponding installation ids
87	as values:
88
89	{"owner/repo_name": inst_id, ...}
90
91	"""
92
93	# Reference:
94	# https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#list-repositories-for-a-user
95
96	results = {}	×
97	for inst_id in get_installation_ids(user_access_token):	×
98	url = f"https://api.github.com/user/installations/{inst_id}/repositories"	×
99	headers = {"Authorization": f"Bearer {user_access_token}"}	×
100	result = curl.get(url, headers=headers)	×
101	doc = RepositoriesResponse.model_validate_json(result.content, strict=True)	×
102	for repo in doc.repositories:	×
103	results[repo.full_name] = inst_id	×
104
105	return results	×
106
107
108	class AccessTokensResponse(BaseModel):	1✔
109	token: str	1✔
110
111
112	def get_installation_access_token(installation_id: int) -> str \| None:	1✔
113	"""Acquire the installation access token for a specific installation id."""
114
115	# Reference:
116	# https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-an-installation-access-token-for-a-github-app
117
118	iat = int(time.time())	×
119	payload = {	×
120	"iat": int(time.time()),
121	"exp": iat + 600,
122	"iss": settings.GITHUB_CLIENT_ID,
123	}
124
125	assert settings.GITHUB_PRIVATE_KEY	×
126	encoded = jwt.encode(payload, settings.GITHUB_PRIVATE_KEY, algorithm="RS256")	×
127	url = f"https://api.github.com/app/installations/{installation_id}/access_tokens"	×
128	result = curl.post(url, headers={"Authorization": f"Bearer {encoded}"})	×
129	if result.status_code == 404:	×
130	# The installation does not exist (our GitHub app has been uninstalled)
131	return None	×
132
133	doc = AccessTokensResponse.model_validate_json(result.content, strict=True)	×
134	return doc.token	×

healthchecks / healthchecks / 24067770611

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous