24067666161

Committed 07 Apr 2026 06:26AM UTC coverage: 91.383%. Remained the same

Build # 24067666161

Build Type

push

github

Committed by

cuu508

Commit Message

Fix log message format in hc.integrations.github.client

Coverage Stats

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

8282 of 9063 relevant lines covered (91.38%)

0.91 hits per line

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

42.62

/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:
        logger.warning(f"Unexpected response from GitHub: {result.content}")

    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	logger.warning(f"Unexpected response from GitHub: {result.content}")	×
69
70	assert doc.installations is not None	×
71	return [item.id for item in doc.installations]	×
72
73
74	class Repository(BaseModel):	1✔
75	full_name: str	1✔
76
77
78	class RepositoriesResponse(BaseModel):	1✔
79	repositories: list[Repository]	1✔
80
81
82	def get_repos(user_access_token: str) -> dict[str, int]:	1✔
83	"""Retrieve the repositories the user has access to.
84
85	Return a dict with repo names as keys and the corresponding installation ids
86	as values:
87
88	{"owner/repo_name": inst_id, ...}
89
90	"""
91
92	# Reference:
93	# https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#list-repositories-for-a-user
94
95	results = {}	×
96	for inst_id in get_installation_ids(user_access_token):	×
97	url = f"https://api.github.com/user/installations/{inst_id}/repositories"	×
98	headers = {"Authorization": f"Bearer {user_access_token}"}	×
99	result = curl.get(url, headers=headers)	×
100	doc = RepositoriesResponse.model_validate_json(result.content, strict=True)	×
101	for repo in doc.repositories:	×
102	results[repo.full_name] = inst_id	×
103
104	return results	×
105
106
107	class AccessTokensResponse(BaseModel):	1✔
108	token: str	1✔
109
110
111	def get_installation_access_token(installation_id: int) -> str \| None:	1✔
112	"""Acquire the installation access token for a specific installation id."""
113
114	# Reference:
115	# https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/generating-an-installation-access-token-for-a-github-app
116
117	iat = int(time.time())	×
118	payload = {	×
119	"iat": int(time.time()),
120	"exp": iat + 600,
121	"iss": settings.GITHUB_CLIENT_ID,
122	}
123
124	assert settings.GITHUB_PRIVATE_KEY	×
125	encoded = jwt.encode(payload, settings.GITHUB_PRIVATE_KEY, algorithm="RS256")	×
126	url = f"https://api.github.com/app/installations/{installation_id}/access_tokens"	×
127	result = curl.post(url, headers={"Authorization": f"Bearer {encoded}"})	×
128	if result.status_code == 404:	×
129	# The installation does not exist (our GitHub app has been uninstalled)
130	return None	×
131
132	doc = AccessTokensResponse.model_validate_json(result.content, strict=True)	×
133	return doc.token	×

healthchecks / healthchecks / 24067666161

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