# fetchcode is a free software tool from nexB Inc. and others.
# Visit https://github.com/aboutcode-org/fetchcode for support and download.
#
# Copyright (c) nexB Inc. and others. All rights reserved.
# http://nexb.com and http://aboutcode.org
#
# This software is licensed under the Apache License version 2.0.
#
# You may not use this software except in compliance with the License.
# You may obtain a copy of the License at:
# http://apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software distributed
# under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
# CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
import hashlib
import os
import sys
from functools import partial
import requests
from dateutil import parser as dateparser
from dateutil.parser import ParserError
def fetch_github_tags_gql(purl):
"""
Yield PackageVersion for given github ``purl`` using the GitHub GQL API.
"""
for node in fetch_github_tag_nodes(purl):
name = node["name"]
target = node["target"]
# in case the tag is a signed tag, then the commit info is in target['target']
if "committedDate" not in target:
target = target["target"]
committed_date = target.get("committedDate")
release_date = None
if committed_date:
try:
release_date = dateparser.parse(committed_date)
except ParserError as e:
pass
yield name, release_date
GQL_QUERY = """
query getTags($name: String!, $owner: String!, $after: String)
{
repository(name: $name, owner: $owner) {
refs(refPrefix: "refs/tags/", first: 100, after: $after) {
totalCount
pageInfo {
endCursor
hasNextPage
}
nodes {
name
target {
... on Commit {
committedDate
}
... on Tag {
target {
... on Commit {
committedDate
}
}
}
}
}
}
}
}"""
def fetch_github_tag_nodes(purl):
"""
Yield node name/target mappings for Git tags of the ``purl``.
Each node has this shape:
{
"name": "v2.6.24-rc5",
"target": {
"target": {
"committedDate": "2007-12-11T03:48:43Z"
}
}
},
"""
variables = {
"owner": purl.namespace,
"name": purl.name,
}
graphql_query = {
"query": GQL_QUERY,
"variables": variables,
}
while True:
response = github_response(graphql_query)
refs = response["data"]["repository"]["refs"]
for node in refs["nodes"]:
yield node
page_info = refs["pageInfo"]
if not page_info["hasNextPage"]:
break
# to fetch next page, we just set the after variable to endCursor
variables["after"] = page_info["endCursor"]
class GitHubTokenError(Exception):
pass
class GraphQLError(Exception):
pass
def get_github_token():
gh_token = os.environ.get("GH_TOKEN", None)
if not gh_token:
from dotenv import load_dotenv
load_dotenv()
gh_token = os.environ.get("GH_TOKEN", None)
return gh_token
def github_response(graphql_query):
gh_token = get_github_token()
if not gh_token:
msg = (
"GitHub API Token Not Set\n"
"Set your GitHub token in the GH_TOKEN environment variable."
)
raise GitHubTokenError(msg)
headers = {"Authorization": f"bearer {gh_token}"}
endpoint = "https://api.github.com/graphql"
response = requests.post(endpoint, headers=headers, json=graphql_query).json()
message = response.get("message")
if message and message == "Bad credentials":
raise GitHubTokenError(f"Invalid GitHub token: {message}")
errors = response.get("errors")
if errors:
raise GraphQLError(errors)
return response
def get_github_rest(url):
headers = None
gh_token = get_github_token()
if gh_token:
headers = {
"Authorization": f"Bearer {gh_token}",
}
return get_response(url, headers)
def get_response(url, headers=None):
resp = requests.get(url, headers=headers)
if resp.status_code == 200:
return resp.json()
raise Exception(f"Failed to fetch: {url}")
def get_text_response(url, headers=None):
resp = requests.get(url, headers=headers)
if resp.status_code == 200:
return resp.text
raise Exception(f"Failed to fetch: {url}")
def make_head_request(url, headers=None):
try:
resp = requests.head(url, headers=headers)
return resp
except requests.RequestException:
raise Exception(f"Failed to fetch: {url}")
def get_hashed_path(name):
"""
Returns a string with a part of the file path derived from the md5 hash.
From https://github.com/CocoaPods/cdn.cocoapods.org:
"There are a set of known prefixes for all Podspec paths, you take the
name of the pod, create a hash (using md5) of it and take the first
three characters."
"""
if not name:
return
podname = get_podname_proper(name)
if name != podname:
name_to_hash = podname
else:
name_to_hash = name
hash_init = get_first_three_md5_hash_characters(name_to_hash)
hashed_path = "/".join(list(hash_init))
return hashed_path
# for FIPS support
sys_v0 = sys.version_info[0]
sys_v1 = sys.version_info[1]
if sys_v0 == 3 and sys_v1 >= 9:
md5_hasher = partial(hashlib.md5, usedforsecurity=False)
else:
md5_hasher = hashlib.md5
def get_podname_proper(podname):
"""
Podnames in cocoapods sometimes are files inside a pods package (like 'OHHTTPStubs/Default')
This returns proper podname in those cases.
"""
if "/" in podname:
return podname.split("/")[0]
return podname
def get_first_three_md5_hash_characters(podname):
"""
From https://github.com/CocoaPods/cdn.cocoapods.org:
"There are a set of known prefixes for all Podspec paths, you take the name of the pod,
create a hash (using md5) of it and take the first three characters."
"""
return md5_hasher(podname.encode("utf-8")).hexdigest()[0:3]
def _http_exists(url: str) -> bool:
"""
Lightweight existence check using a ranged GET so CDNs/servers that ignore HEAD still work.
"""
try:
resp = make_head_request(url, headers={"Range": "bytes=0-0"})
return resp is not None and resp.status_code in (200, 206)
except Exception:
return False