tmp: bootstrap parsing woodpecker logs with stolen enough code

infrastructure/woodpecker.py

@@ -0,0 +1,98 @@
+# Copyright © 2022 Aravinth Manivannan <realaravinth@batsense.net>
+# Copyright © 2022 enough.community https://lab.enough.community/main/infrastructure/-/blob/master/AUTHORS
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+from urillib.parse import urlparse, urlunparse
+
+from django.conf import settings
+
+
+class CI:
+    def __init__(self):
+        self.host = settings.HOSTEA["META"]["WOODPECKER"]["HOST"]
+        # checking if woodpecker host is URL
+        _ = urlparse(self.host)
+        token = settings.HOSTEA["META"]["WOODPECKER"]["TOKEN"]
+        self.auth_header = {"Authorization": f"Bearer {token}"}
+
+        self.w = requests.Session()
+        self.w.url = f"https://{self.host}"
+        r = self.w.get(self.w.url + "/authorize", allow_redirects=False)
+
+        location = self.gitea_browser.confirm_oauth(
+            r.headers["Location"], f"https://{self.hostname}/authorize"
+        )
+        r = self.w.get(location, allow_redirects=False)
+        r.raise_for_status()
+
+        #
+        # Woodpecker CSRF
+        #
+        r = self.w.get(self.w.url + "/web-config.js", allow_redirects=False)
+        r.raise_for_status()
+        csrf = re.findall('window.WOODPECKER_CSRF = "(.*?)"', r.text)[0]
+
+        #
+        # Woodpecker token
+        #
+        r = self.w.post(
+            self.w.url + "/api/user/token",
+            headers={"X-CSRF-TOKEN": csrf},
+            allow_redirects=False,
+        )
+        r.raise_for_status()
+        self.token = r.text
+        self.w.headers = {"Authorization": f"Bearer {self.token}"}
+
+    def confirm_oauth(self, url, redirect):
+        logger.info(f"confirm oauth {url} redirect {redirect}")
+        r = self.g.get(url, allow_redirects=False)
+        r.raise_for_status()
+        if r.status_code == 200:
+            soup = BeautifulSoup(r.text, "html.parser")
+            data = {
+                "redirect_uri": redirect,
+            }
+            for input in soup.select(
+                'form[action="/login/oauth/grant"] input[type="hidden"]'
+            ):
+                if (
+                    input.get("name") is None
+                    or input.get("value") is None
+                    or input.get("value") == ""
+                ):
+                    continue
+                logger.info(f"collected hidden input {input['name']} {input['value']}")
+                data[input["name"]] = input["value"]
+            assert len(data) > 1, f"{data} has only one field, more are expected"
+            r = self.g.post(
+                self.g.url + "/login/oauth/grant", data=data, allow_redirects=False
+            )
+            r.raise_for_status()
+            logger.info("oauth confirmed")
+        elif r.status_code == 302:
+            logger.info("no confirmation required")
+        location = r.headers["Location"]
+        logger.info(f"going back to {location}")
+        assert location.startswith(redirect)
+        return location
+
+    def _build_list_url(self):
+        parsed = urlparse(self.host)
+        list_builds = "/api/repos/Hostea/dashboard/builds"
+        return (urlunparse((parsed.scheme, parsed.netloc, list_builds, "", "", "")),)
+
+    def logs(self):
+        resp = request.get(self._build_list_url(), headers=self.auth_header)
+        print(resp.json())