99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
|
# 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())
|