You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
193 lines
6.3 KiB
193 lines
6.3 KiB
from swagger_client.api.admin_api import AdminApi
|
|
from swagger_client.api.user_api import UserApi
|
|
from swagger_client.api.organization_api import OrganizationApi
|
|
from swagger_client.api_client import ApiClient
|
|
from swagger_client.models import CreateUserOption, Team, User, Organization
|
|
from swagger_client.rest import ApiException
|
|
from sys import stderr
|
|
import click
|
|
import termcolor
|
|
import requests
|
|
from bs4 import BeautifulSoup as bs
|
|
import random
|
|
import string
|
|
|
|
|
|
# load configuration
|
|
try:
|
|
import config
|
|
except ImportError:
|
|
import config_default as config
|
|
|
|
HOST = config.HOST
|
|
ADMIN_TOKEN = config.ADMIN_TOKEN
|
|
|
|
|
|
def rand_str(n) -> str:
|
|
return ''.join([random.choice(string.ascii_lowercase) for _ in range(n)])
|
|
|
|
|
|
def text_red(text) -> str:
|
|
return termcolor.colored(text, 'red')
|
|
|
|
|
|
def text_green(text) -> str:
|
|
return termcolor.colored(text, 'green')
|
|
|
|
|
|
def text_orange(text) -> str:
|
|
return termcolor.colored(text, 'yellow')
|
|
|
|
|
|
def does_user_exist(user_api: UserApi, username: str) -> bool:
|
|
try:
|
|
user = user_api.user_get(username)
|
|
return user is not None
|
|
except ApiException as e:
|
|
if e.status == 404:
|
|
# uzivatel nebyl nalezen
|
|
return False
|
|
else:
|
|
raise e
|
|
|
|
|
|
def does_organization_exist(org_api: OrganizationApi, orgname: str) -> bool:
|
|
try:
|
|
org = org_api.org_get(orgname)
|
|
return org is not None
|
|
except ApiException as e:
|
|
if e.status == 404:
|
|
# organizace nebyla nalezena
|
|
return False
|
|
else:
|
|
raise e
|
|
|
|
|
|
def does_organization_have_org_team(org_api: OrganizationApi, orgname: str) -> bool:
|
|
teams = org_api.org_list_teams(orgname)
|
|
for t in teams:
|
|
if 'org' == t.name:
|
|
return True
|
|
return False
|
|
|
|
|
|
def get_org_team_id(org_api: OrganizationApi, orgname: str) -> int:
|
|
teams = org_api.org_list_teams(orgname)
|
|
for t in teams:
|
|
if 'org' == t.name:
|
|
return t.id
|
|
raise AssertionError('Tento kod mel byt unreachable. Zmizel tym orgu za behu skriptu?')
|
|
|
|
|
|
def validate(func_check, message: str):
|
|
print(message, end="")
|
|
try:
|
|
if func_check():
|
|
print(text_green(' [OK]'))
|
|
return
|
|
except ApiException as e:
|
|
print(text_red(' [FAIL]'))
|
|
print(text_red(e.body))
|
|
print("Request se nepodaril! Duvod:", e.reason, file=stderr)
|
|
exit(1)
|
|
|
|
print(text_red(' [FAIL]'))
|
|
print("Neocekavany stav systemu!", file=stderr)
|
|
exit(1)
|
|
|
|
|
|
def checked_api_action(func_action, message: str):
|
|
print(message, end=' ')
|
|
try:
|
|
res = func_action()
|
|
print(text_green('[OK]'))
|
|
return res
|
|
except ApiException as e:
|
|
print(text_red('[FAIL]'))
|
|
print(text_red('Server odpovedel:'))
|
|
print(text_red(e.body))
|
|
exit(1)
|
|
|
|
|
|
def reset_password(email: str):
|
|
url = HOST.replace('api/v1', 'user/forgot_password')
|
|
with requests.session() as session:
|
|
# ziskani formulare
|
|
resp = session.get(url)
|
|
|
|
# vyplneni formulare
|
|
soup = bs(resp.text, features='lxml')
|
|
form = {n['name']: n['value'] for n in soup.find_all('input')}
|
|
form['email'] = email
|
|
|
|
# submit form
|
|
resp = session.post(url, data=form)
|
|
|
|
# simulate being standard api call
|
|
if resp.status_code != 200:
|
|
raise ApiException(status=resp.status_code, reason=resp.reason, http_resp='Form submission failed.')
|
|
|
|
|
|
@click.command()
|
|
@click.option('-a', '--admin-token', help='Admin access token')
|
|
@click.option('-h', '--host', help='Gitea URL (default: {})'.format(HOST))
|
|
@click.option('-e', '--email', help='Emailova adresa noveho uzivatele.', required=True)
|
|
@click.option('-u', '--username', help='Uzivatelske jmeno noveho uzivatele.', required=True)
|
|
@click.option('-f', '--fullname', help='Plne jmeno noveho uzivatele.')
|
|
@click.option('-s', '--seminar', help='Seminar, kde uzivatel orguje.', required=True)
|
|
def register(username: str, seminar: str, fullname: str, email: str, host, admin_token):
|
|
"""
|
|
Nastroj pro automatizaci zakladani uctu na serveru Gitea.
|
|
|
|
CO SKRIPT DELA:
|
|
|
|
Prvni probehne kontrola stavu systemu, zda je mozne uzivatele bezpecne zalozit. Pote se provede pokus o zalozeni
|
|
nasledovan pridanim uzivatele do skupiny 'org' v pozadovane organizaci (jmeno dle seminare).
|
|
|
|
Vytvoreny uzivatel ma nastavene nahodne heslo. To je automaticky resetovano a o tom je odeslan email.
|
|
|
|
KONFIGURACE:
|
|
|
|
V souboru config.py je mozne nastavit pristupovy token a URL perzistentne. Pristupovy token musi byt od uzivatele,
|
|
ktery je ve skupine 'Owners' v pozadovanem seminari. Jinak nema dostatecna prava na pridavani uzivatelu.
|
|
"""
|
|
|
|
global HOST
|
|
HOST = host if host is not None else HOST
|
|
|
|
global ADMIN_TOKEN
|
|
ADMIN_TOKEN = admin_token if admin_token is not None else ADMIN_TOKEN
|
|
|
|
api_client = ApiClient()
|
|
api_client.configuration.host = host if host is not None else HOST
|
|
api_client.configuration.api_key = {'token': admin_token if admin_token is not None else ADMIN_TOKEN}
|
|
|
|
admin_api = AdminApi(api_client)
|
|
user_api = UserApi(api_client)
|
|
org_api = OrganizationApi(api_client)
|
|
|
|
# kontrola predpokladu o stavu systemu
|
|
validate(lambda: not does_user_exist(user_api, username), "Uzivatelske jmeno je volne...")
|
|
validate(lambda: does_organization_exist(org_api, seminar), "Organizace pro seminar existuje...")
|
|
validate(lambda: does_organization_have_org_team(org_api, seminar), "V danem seminari existuje team 'org'...")
|
|
print(text_green('Zakladni predpoklady pro uspesne zalozeni uctu splneny!\n'))
|
|
|
|
# nasbirame vsechny potrebne informace
|
|
org_team_id = get_org_team_id(org_api, seminar)
|
|
|
|
# pripravime si uzivatele na zalozeni
|
|
create_user_req = CreateUserOption(email=email, full_name=fullname, username=username,
|
|
password=rand_str(20))
|
|
|
|
# provedeme zmeny v systemu
|
|
user = checked_api_action(lambda: admin_api.admin_create_user(body=create_user_req), "Zakladam noveho uzivatele...")
|
|
checked_api_action(lambda: org_api.org_add_team_member(org_team_id, username),
|
|
"Pridavam do teamu 'org' v seminari {}...".format(seminar))
|
|
print(text_green('Uzivatel zalozen!'))
|
|
print()
|
|
|
|
checked_api_action(lambda: reset_password(email), "Resetuji heslo...")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
register()
|
|
|