Skip to content

security: harden contribute.py with token validation, error sanitization, and retry/backoff (fixes #288) #398

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pradeeban merged 3 commits into from
Feb 20, 2026
Merged

security: harden contribute.py with token validation, error sanitization, and retry/backoff (fixes #288) #398

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fb5b10e
security: harden contribute.py with token validation, safe error hand…
GaneshPatil7517 Feb 18, 2026
9ace755
security: address PR review - fix exception types to github.GithubExc…
GaneshPatil7517 Feb 18, 2026
08bd61b
refactor: remove Fix N counter prefixes from comments per review feed…
GaneshPatil7517 Feb 19, 2026
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
374 changes: 213 additions & 161 deletions contribute.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,162 +1,214 @@
import github
from github import Github
import os,sys,platform,base64,time

# Intializing the Variables
BOT_TOKEN = os.environ.get('CONCORE_BOT_TOKEN', '')
BOT_ACCOUNT = 'concore-bot' #bot account name
REPO_NAME = 'concore-studies' #study repo name
UPSTREAM_ACCOUNT = 'ControlCore-Project' #upstream account name
STUDY_NAME = sys.argv[1]
STUDY_NAME_PATH = sys.argv[2]
AUTHOR_NAME = sys.argv[3]
BRANCH_NAME = sys.argv[4]
PR_TITLE = sys.argv[5]
PR_BODY = sys.argv[6]

# Defining Functions
def checkInputValidity():
if not AUTHOR_NAME or not STUDY_NAME or not STUDY_NAME_PATH:
print("Please Provide necessary Inputs")
exit(1)
if not os.path.isdir(STUDY_NAME_PATH):
print("Directory doesnot Exists.Invalid Path")
exit(1)

def printPR(pr):
print(f'Check your example here https://github.com/{UPSTREAM_ACCOUNT}/{REPO_NAME}/pulls/{pr.number}',end="")

def anyOpenPR(upstream_repo):
try:
prs = upstream_repo.get_pulls(state='open', head=f'{BOT_ACCOUNT}:{BRANCH_NAME}')
return prs[0] if prs.totalCount > 0 else None
except Exception:
print("Unable to fetch PR status. Try again later.")
exit(1)

def commitAndUpdateRef(repo,tree_content,commit,branch):
try:
new_tree = repo.create_git_tree(tree=tree_content,base_tree=commit.commit.tree)
new_commit = repo.create_git_commit(f"Committing Study Named {STUDY_NAME}",new_tree,[commit.commit])
if len(repo.compare(base=commit.commit.sha,head=new_commit.sha).files) == 0:
print("Your don't have any new changes.May be your example is already accepted.If this is not the case try with different fields.")
exit(1)
ref = repo.get_git_ref("heads/"+branch.name)
ref.edit(new_commit.sha,True)
except Exception as e:
print("failed to Upload your example.Please try after some time.",end="")
exit(1)


def appendBlobInTree(repo,content,file_path,tree_content):
blob = repo.create_git_blob(content,'utf-8')
tree_content.append( github.InputGitTreeElement(path=file_path,mode="100644",type="blob",sha=blob.sha))


def runWorkflow(repo,upstream_repo):
openPR = anyOpenPR(upstream_repo)
if not openPR:
try:
repo.get_workflow("pull_request.yml").create_dispatch(
ref=BRANCH_NAME,
inputs={'title': f"[BOT]: {PR_TITLE}", 'body': PR_BODY, 'upstreamRepo': UPSTREAM_ACCOUNT, 'botRepo': BOT_ACCOUNT, 'repo': REPO_NAME}
)
printPRStatus(upstream_repo)
except Exception as e:
print(f"Error triggering workflow. Try again later.\n ERROR: {e}")
exit(1)
else:
print(f"Successfully uploaded. Waiting for approval: https://github.com/{UPSTREAM_ACCOUNT}/{REPO_NAME}/pull/{openPR.number}")

def printPRStatus(upstream_repo):
attempts = 5
delay = 2
for i in range(attempts):
print(f"Attempt: {i}")
try:
latest_pr = upstream_repo.get_pulls(state='open', sort='created', direction='desc')[0]
print(f"Check your example here: https://github.com/{UPSTREAM_ACCOUNT}/{REPO_NAME}/pull/{latest_pr.number}")
return
except Exception:
time.sleep(delay)
delay *= 2
print("Uploaded successfully, but unable to fetch status.")


def isImageFile(filename):
image_extensions = ['.jpeg', '.jpg', '.png','.gif']
return any(filename.endswith(ext) for ext in image_extensions)

def remove_prefix(text, prefix):
if text.startswith(prefix):
return text[len(prefix):]
return text


# Decode Github Token
def decode_token(encoded_token):
decoded_bytes = encoded_token.encode("ascii")
convertedbytes = base64.b64decode(decoded_bytes)
decoded_token = convertedbytes.decode("ascii")
return decoded_token


# check if directory path is Valid
checkInputValidity()


# Authenticating Github with Access token
try:
BRANCH_NAME = AUTHOR_NAME.replace(" ", "_") + "_" + STUDY_NAME if BRANCH_NAME == "#" else BRANCH_NAME.replace(" ", "_")
PR_TITLE = f"Contributing Study {STUDY_NAME} by {AUTHOR_NAME}" if PR_TITLE == "#" else PR_TITLE
PR_BODY = f"Study Name: {STUDY_NAME}\nAuthor Name: {AUTHOR_NAME}" if PR_BODY == "#" else PR_BODY
DIR_PATH = STUDY_NAME
DIR_PATH = DIR_PATH.replace(" ","_")
g = Github(BOT_TOKEN)
repo = g.get_user(BOT_ACCOUNT).get_repo(REPO_NAME)
upstream_repo = g.get_repo(f'{UPSTREAM_ACCOUNT}/{REPO_NAME}') #controlcore-Project/concore-studies
base_ref = upstream_repo.get_branch(repo.default_branch)

try:
repo.get_branch(BRANCH_NAME)
is_present = True
except github.GithubException:
print(f"No Branch is available with the name {BRANCH_NAME}")
is_present = False
except Exception as e:
print("Authentication failed", end="")
exit(1)


try:
if not is_present:
repo.create_git_ref(f"refs/heads/{BRANCH_NAME}", base_ref.commit.sha)
branch = repo.get_branch(BRANCH_NAME)
except Exception:
print("Unable to create study. Try again later.")
exit(1)


tree_content = []

try:
for root, dirs, files in os.walk(STUDY_NAME_PATH):
files = [f for f in files if not f[0] == '.']
for filename in files:
path = f"{root}/{filename}"
if isImageFile(filename):
with open(file=path, mode='rb') as file:
image = file.read()
content = base64.b64encode(image).decode('utf-8')
else:
with open(file=path, mode='r') as file:
content = file.read()
file_path = f'{DIR_PATH+remove_prefix(path,STUDY_NAME_PATH)}'
if(platform.uname()[0]=='Windows'): file_path=file_path.replace("\\","/")
appendBlobInTree(repo,content,file_path,tree_content)
commitAndUpdateRef(repo,tree_content,base_ref.commit,branch)
runWorkflow(repo,upstream_repo)
except Exception as e:
print(e)
print("Some error Occured.Please try again after some time.",end="")
import github
from github import Github
import os,sys,platform,base64,time,re
import requests
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated

# Intializing the Variables
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
BOT_TOKEN = os.environ.get('CONCORE_BOT_TOKEN', '')

# Fix 1: Fail fast if token is missing

@pradeeban pradeeban Feb 19, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you remove the "Fix 1" "Fix 2" "Fix 3" ... counters please?
Just noting, "Fail fast if token is missing" is sufficient.

The counter of errors indeed help with reviewing the PR. But it shouldn't live in the code itself.

@GaneshPatil7517 GaneshPatil7517 Feb 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok @pradeeban sir ill start working on it immediately....

@GaneshPatil7517 GaneshPatil7517 Feb 19, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you for the feedback, @pradeeban Sir You're absolutely right the "Fix 1:", "Fix 2:", etc. counters were meant for PR traceability but don't belong in the code itself.
I've removed all the counter prefixes and kept only the descriptive comments (e.g., # Fail fast if token is missing). Changes pushed in commit 08bd61b.
Please let me know if there's anything else to address.

if not BOT_TOKEN:
print("Error: CONCORE_BOT_TOKEN environment variable is not set.")
sys.exit(1)

# Fix 2: Token format validation
token_pattern = r"^(ghp_|github_pat_)[A-Za-z0-9_]{20,}$"
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
if not re.match(token_pattern, BOT_TOKEN):
print("Error: Invalid GitHub token format.")
sys.exit(1)
BOT_ACCOUNT = 'concore-bot' #bot account name
REPO_NAME = 'concore-studies' #study repo name
UPSTREAM_ACCOUNT = 'ControlCore-Project' #upstream account name
STUDY_NAME = sys.argv[1]
STUDY_NAME_PATH = sys.argv[2]
AUTHOR_NAME = sys.argv[3]
BRANCH_NAME = sys.argv[4]
PR_TITLE = sys.argv[5]
PR_BODY = sys.argv[6]

# Defining Functions
def checkInputValidity():
if not AUTHOR_NAME or not STUDY_NAME or not STUDY_NAME_PATH:
print("Please Provide necessary Inputs")
exit(1)
if not os.path.isdir(STUDY_NAME_PATH):
print("Directory doesnot Exists.Invalid Path")
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
exit(1)

# Fix 5: Retry + backoff wrapper for GitHub API requests
def github_request(method, url, headers=None, json=None, retries=3):
for attempt in range(retries):
try:
response = requests.request(method, url, headers=headers, json=json, timeout=30)
if response.status_code == 429 or response.status_code >= 500:
wait_time = 2 ** attempt
time.sleep(wait_time)
continue
return response
except requests.exceptions.ConnectionError:
print("Network error while contacting GitHub API.")
sys.exit(1)
except requests.exceptions.Timeout:
print("GitHub API request timed out.")
sys.exit(1)
print("Error: GitHub API request failed after retries.")
sys.exit(1)

Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
# Fix 4: Correct PR URL (singular 'pull' not 'pulls')
def printPR(pr):
print(f'Check your example here https://github.com/{UPSTREAM_ACCOUNT}/{REPO_NAME}/pull/{pr.number}',end="")

def anyOpenPR(upstream_repo):
try:
prs = upstream_repo.get_pulls(state='open', head=f'{BOT_ACCOUNT}:{BRANCH_NAME}')
return prs[0] if prs.totalCount > 0 else None
except requests.exceptions.ConnectionError:
print("Network error while fetching PR status.")
exit(1)
except requests.exceptions.Timeout:
print("Request timed out while fetching PR status.")
exit(1)
except Exception:
print("Unable to fetch PR status. Try again later.")
exit(1)
Comment thread
GaneshPatil7517 marked this conversation as resolved.

def commitAndUpdateRef(repo,tree_content,commit,branch):
try:
new_tree = repo.create_git_tree(tree=tree_content,base_tree=commit.commit.tree)
new_commit = repo.create_git_commit(f"Committing Study Named {STUDY_NAME}",new_tree,[commit.commit])
if len(repo.compare(base=commit.commit.sha,head=new_commit.sha).files) == 0:
print("Your don't have any new changes.May be your example is already accepted.If this is not the case try with different fields.")
exit(1)
ref = repo.get_git_ref("heads/"+branch.name)
ref.edit(new_commit.sha,True)
except requests.exceptions.HTTPError as e:
print(f"GitHub API error: {e.response.status_code}")
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
exit(1)
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
except Exception:
print("Failed to upload your example. Please try after some time.",end="")
exit(1)


def appendBlobInTree(repo,content,file_path,tree_content):
blob = repo.create_git_blob(content,'utf-8')
tree_content.append( github.InputGitTreeElement(path=file_path,mode="100644",type="blob",sha=blob.sha))


def runWorkflow(repo,upstream_repo):
openPR = anyOpenPR(upstream_repo)
if not openPR:
try:
repo.get_workflow("pull_request.yml").create_dispatch(
ref=BRANCH_NAME,
inputs={'title': f"[BOT]: {PR_TITLE}", 'body': PR_BODY, 'upstreamRepo': UPSTREAM_ACCOUNT, 'botRepo': BOT_ACCOUNT, 'repo': REPO_NAME}
)
printPRStatus(upstream_repo)
except requests.exceptions.HTTPError as e:
print(f"GitHub API error while triggering workflow: {e.response.status_code}")
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
exit(1)
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
except Exception:
print("Error triggering workflow. Try again later.")
exit(1)
else:
print(f"Successfully uploaded. Waiting for approval: https://github.com/{UPSTREAM_ACCOUNT}/{REPO_NAME}/pull/{openPR.number}")

def printPRStatus(upstream_repo):
attempts = 5
delay = 2
for i in range(attempts):
print(f"Attempt: {i}")
try:
latest_pr = upstream_repo.get_pulls(state='open', sort='created', direction='desc')[0]
print(f"Check your example here: https://github.com/{UPSTREAM_ACCOUNT}/{REPO_NAME}/pull/{latest_pr.number}")
return
except Exception:
time.sleep(delay)
delay *= 2
print("Uploaded successfully, but unable to fetch status.")


def isImageFile(filename):
image_extensions = ['.jpeg', '.jpg', '.png','.gif']
return any(filename.endswith(ext) for ext in image_extensions)

def remove_prefix(text, prefix):
if text.startswith(prefix):
return text[len(prefix):]
return text


# Fix 9: Removed unused decode_token() function

# check if directory path is Valid
checkInputValidity()


# Authenticating Github with Access token
try:
BRANCH_NAME = AUTHOR_NAME.replace(" ", "_") + "_" + STUDY_NAME if BRANCH_NAME == "#" else BRANCH_NAME.replace(" ", "_")
PR_TITLE = f"Contributing Study {STUDY_NAME} by {AUTHOR_NAME}" if PR_TITLE == "#" else PR_TITLE
PR_BODY = f"Study Name: {STUDY_NAME}\nAuthor Name: {AUTHOR_NAME}" if PR_BODY == "#" else PR_BODY
DIR_PATH = STUDY_NAME
DIR_PATH = DIR_PATH.replace(" ","_")
g = Github(BOT_TOKEN)
repo = g.get_user(BOT_ACCOUNT).get_repo(REPO_NAME)
upstream_repo = g.get_repo(f'{UPSTREAM_ACCOUNT}/{REPO_NAME}') #controlcore-Project/concore-studies
base_ref = upstream_repo.get_branch(repo.default_branch)

try:
repo.get_branch(BRANCH_NAME)
is_present = True
except github.GithubException:
print(f"No Branch is available with the name {BRANCH_NAME}")
is_present = False
except requests.exceptions.ConnectionError:
print("Network error during GitHub authentication.")
exit(1)
except requests.exceptions.Timeout:
print("GitHub authentication request timed out.")
exit(1)
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
except Exception:
print("Authentication failed", end="")
exit(1)


try:
if not is_present:
repo.create_git_ref(f"refs/heads/{BRANCH_NAME}", base_ref.commit.sha)
branch = repo.get_branch(BRANCH_NAME)
except Exception:
print("Unable to create study. Try again later.")
exit(1)


tree_content = []

try:
for root, dirs, files in os.walk(STUDY_NAME_PATH):
files = [f for f in files if not f[0] == '.']
for filename in files:
path = f"{root}/{filename}"
if isImageFile(filename):
with open(file=path, mode='rb') as file:
image = file.read()
content = base64.b64encode(image).decode('utf-8')
else:
with open(file=path, mode='r') as file:
content = file.read()
file_path = f'{DIR_PATH+remove_prefix(path,STUDY_NAME_PATH)}'
if(platform.uname()[0]=='Windows'): file_path=file_path.replace("\\","/")
appendBlobInTree(repo,content,file_path,tree_content)
commitAndUpdateRef(repo,tree_content,base_ref.commit,branch)
runWorkflow(repo,upstream_repo)
except requests.exceptions.HTTPError as e:
print(f"GitHub API error: {e.response.status_code}")
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
exit(1)
except requests.exceptions.ConnectionError:
print("Network error while uploading study.")
exit(1)
except requests.exceptions.Timeout:
print("Request timed out while uploading study.")
exit(1)
Comment thread
GaneshPatil7517 marked this conversation as resolved.
Outdated
except Exception:
print("Some error occurred. Please try again after some time.",end="")
exit(1)
Loading