Page MenuHomePhorge
Files F85804686

No OneTemporary
Actions

Size
4 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lilybuild/lilybuild/auth.py b/lilybuild/lilybuild/auth.py
index cd3f58f..cf00a48 100644
--- a/lilybuild/lilybuild/auth.py
+++ b/lilybuild/lilybuild/auth.py
@@ -1,51 +1,113 @@
from buildbot.plugins import *
from buildbot.www.oauth2 import OAuth2Auth
-from buildbot.www.authz.roles import RolesFromBase
+from buildbot.www.authz.roles import RolesFromBase, RolesFromOwner
+from buildbot.www.authz.authz import Authz
+from buildbot.data.builds import BuildEndpoint
+from twisted.internet import defer
+from twisted.python import log
+import requests
+import txrequests
class ForgejoAuth(OAuth2Auth):
name = 'Forgejo'
faIcon = 'fa-git'
def __init__(self, instanceUri, clientId, clientSecret, **kwargs):
uri = instanceUri.rstrip('/')
self.authUri = f'{uri}/login/oauth/authorize'
self.tokenUri = f'{uri}/login/oauth/access_token'
self.resourceEndpoint = f'{uri}/api/v1'
super().__init__(clientId, clientSecret, **kwargs)
def getUserInfoFromOAuthClient(self, c):
+ access_token = c.params['access_token']
user = self.get(c, '/user')
groups = self.get(c, '/user/orgs')
return {
+ 'access_token': access_token,
'full_name': user['full_name'],
'username': user['login'],
'email': user['email'],
'avatar_url': user['avatar_url'],
'groups': [g['name'] for g in groups],
'is_admin': user['is_admin'],
}
class RolesFromGroups(RolesFromBase):
def __init__(self, prefix=''):
super().__init__()
self.prefix = prefix
def getRolesFromUser(self, userDetails):
roles = []
if 'groups' in userDetails:
for group in userDetails['groups']:
roles.append(self.prefix + group)
return roles
class RolesFromAttr(RolesFromBase):
def __init__(self, attr_name, role_name):
super().__init__()
self.attr_name = attr_name
self.role_name = role_name
def getRolesFromUser(self, userDetails):
if userDetails.get(self.attr_name):
return [self.role_name]
else:
return []
+
+class ForgejoAuthz(Authz):
+ def __init__(self, base_url, access_token, **kwargs):
+ self.forgejo_base_url = base_url.rstrip('/') + '/'
+ self.forgejo_api_url = self.forgejo_base_url + 'api/v1'
+ self.forgejo_access_token = access_token
+ self.session = txrequests.Session()
+ super().__init__(**kwargs)
+
+ @defer.inlineCallbacks
+ def assertUserAllowed(self, ep, action, options, userDetails):
+ try:
+ (epobject, epdict) = self.master.data.getEndpoint(ep)
+ if isinstance(epobject, BuildEndpoint):
+ build_props = yield self.master.data.get(('builds', epdict['buildid'], 'properties'))
+ ok = yield self.user_can_access_build(userDetails, build_props)
+ if ok:
+ return defer.succeed(None)
+ except Exception as e:
+ print('exception:', e)
+ finally:
+ pass
+ res = yield super().assertUserAllowed(ep, action, options, userDetails)
+ return res
+
+ @defer.inlineCallbacks
+ def user_can_access_build(self, userDetails, build_props):
+ if 'lilybuild_repo' not in build_props:
+ return False
+ repo = build_props['lilybuild_repo'][0]
+ if not repo.startswith(self.forgejo_base_url):
+ return False
+ repo_name = repo[len(self.forgejo_base_url):]
+ if repo_name.count('/') != 1:
+ return False
+
+ username = userDetails['username']
+ access_token = userDetails['access_token']
+
+ # https://codeberg.org/forgejo/forgejo/issues/6837
+ # If I am not the repo owner, this works
+ resp = yield self.session.get(
+ f'{self.forgejo_api_url}/repos/{repo_name}/collaborators/{username}',
+ headers={'Authorization': f'token {access_token}'}
+ )
+ if resp.status_code < 300:
+ return True
+
+ # If I am the repo owner, this works
+ resp = yield self.session.get(
+ f'{self.forgejo_api_url}/repos/{repo_name}/collaborators/{username}/permission',
+ headers={'Authorization': f'token {access_token}'}
+ )
+ return resp.status_code < 300

File Metadata

Mime Type
text/x-diff
Expires
Sat, Oct 10, 12:54 PM (1 d, 7 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1785215
Default Alt Text
(4 KB)

Event Timeline