-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
examples: github webhook event logger: Initial commit
Related: https://gist.github.com/2bb4bb6d7a6abaa07cebc7c04d1cafa5 Related: https://github.com/pdxjohnny/scitt-api-emulator/blob/7ec55acd2b09bb2e08df9c56aa90cfb431aa3fd2/docs/slsa_in_toto.md Signed-off-by: John Andersen <[email protected]>
- Loading branch information
Showing
1 changed file
with
90 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,90 @@ | ||
""" | ||
Usage | ||
***** | ||
.. code-block:: console | ||
$ python github_webhook_event_logger.py -R org/repo | ||
""" | ||
import os | ||
import sys | ||
import json | ||
import pathlib | ||
import argparse | ||
import subprocess | ||
import http.server | ||
|
||
import httptest | ||
|
||
|
||
class GitHubWebhookLogger(http.server.BaseHTTPRequestHandler): | ||
def do_POST(self): | ||
self.send_response(200) | ||
self.send_header("Content-type", "text/plain") | ||
self.send_header("Content-length", 3) | ||
self.end_headers() | ||
self.wfile.write(b"OK\n") | ||
|
||
payload = json.loads(self.rfile.read()) | ||
payload_path = pathlib.Path( | ||
".".join( | ||
[ | ||
self.headers["X-github-event"], | ||
*( | ||
[ | ||
":".join([key, value]) | ||
for key, value in payload.items() | ||
if isinstance(value, str) and not os.sep in value | ||
] | ||
+ ["json"] | ||
), | ||
], | ||
) | ||
) | ||
|
||
payload_path.write_text(json.dumps(payload, indent=4, sort_keys=True)) | ||
|
||
@classmethod | ||
def cli(cls, argv=None): | ||
parser = argparse.ArgumentParser(description="Log GitHub webhook events") | ||
parser.add_argument( | ||
"-R", | ||
"--repo", | ||
dest="org_and_repo", | ||
required=True, | ||
type=str, | ||
help="GitHub repo in format org/repo", | ||
) | ||
parser.add_argument( | ||
"--state-dir", | ||
dest="state_dir", | ||
type=pathlib.Path, | ||
help="Directory to cache requests in", | ||
default=pathlib.Path(os.getcwd(), ".cache", "httptest"), | ||
) | ||
|
||
args = parser.parse_args(argv) | ||
|
||
if not args.state_dir.is_dir(): | ||
args.state_dir.mkdir(parents=True) | ||
|
||
with httptest.Server(cls) as logging_server: | ||
with httptest.Server( | ||
httptest.CachingProxyHandler.to( | ||
logging_server.url(), state_dir=str(args.state_dir.resolve()) | ||
) | ||
) as cache_server: | ||
subprocess.check_call( | ||
[ | ||
"gh", | ||
"webhook", | ||
"forward", | ||
f"--repo={args.org_and_repo}", | ||
"--events=*", | ||
f"--url={cache_server.url()}", | ||
] | ||
) | ||
|
||
|
||
if __name__ == "__main__": | ||
GitHubWebhookLogger.cli(sys.argv[1:]) |