import logging import os import re import configparser from pathlib import Path from werkzeug.security import check_password_hash from flask import Flask, render_template, request, redirect, flash, safe_join, send_file, jsonify, url_for from flask_dropzone import Dropzone from flask_migrate import Migrate, MigrateCommand from flask_script import Manager from flask_login import LoginManager, login_required, login_user, current_user, logout_user from portal.model import db, RegisteredUser import portal.solar # TODO: Integrate Shibboleth login. # TODO: make logging level configurable logging.basicConfig(level=logging.DEBUG, format='[APP LOGGER] %(asctime)s %(levelname)s: %(message)s') ###################### # Load configuration # ###################### config = configparser.ConfigParser() config.read('config.ini') config = config['DEFAULT'] MAIL_HOST = config['MAIL_HOST'] MAIL_LOGIN = config['MAIL_LOGIN'] MAIL_PASS = config['MAIL_PASS'] APP_SECRET_KEY = bytes.fromhex(config['APP_SECRET_KEY']) SMTP_PORT = int(config['SMTP_PORT']) IMAP_PORT = int(config['IMAP_PORT']) MAX_UPLOAD_SIZE = int(config['MAX_UPLOAD_SIZE']) # Bytes MAX_FILES_PER_UPLOAD = int(config['MAX_FILES_PER_UPLOAD']) CONTRACT_CLIENT_CONTACT = config['CONTRACT_CLIENT_CONTACT'] MAIL_SUBJECT = config['MAIL_SUBJECT'] MAIL_BODY = config['MAIL_BODY'] SQL_CONN_STR = config['SQL_CONN_STR'] if 'UPLOADS_DIR' in config: UPLOADS_DIR = Path(config['UPLOADS_DIR']) else: UPLOADS_DIR = Path(__file__).resolve().parent / 'uploads' if not UPLOADS_DIR.exists: UPLOADS_DIR.mkdir(parents=True) # Override configs with environment variables, if set if 'PORTALDS4DS1_MAIL_HOST' in os.environ: MAIL_HOST = os.environ['PORTALDS4DS1_MAIL_HOST'] if 'PORTALDS4DS1_MAIL_LOGIN' in os.environ: MAIL_LOGIN = os.environ['PORTALDS4DS1_MAIL_LOGIN'] if 'PORTALDS4DS1_MAIL_PASS' in os.environ: MAIL_PASS = os.environ['PORTALDS4DS1_MAIL_PASS'] if 'PORTALDS4DS1_APP_SECRET_KEY' in os.environ: APP_SECRET_KEY = bytes.fromhex(os.environ['PORTALDS4DS1_APP_SECRET_KEY']) if 'PORTALDS4DS1_SMTP_PORT' in os.environ: SMTP_PORT = int(os.environ['PORTALDS4DS1_SMTP_PORT']) if 'PORTALDS4DS1_IMAP_PORT' in os.environ: IMAP_PORT = int(os.environ['PORTALDS4DS1_IMAP_PORT']) if 'PORTALDS4DS1_MAX_UPLOAD_SIZE' in os.environ: MAX_UPLOAD_SIZE = int(os.environ['PORTALDS4DS1_MAX_UPLOAD_SIZE']) if 'PORTALDS4DS1_MAX_FILES_PER_UPLOAD' in os.environ: MAX_FILES_PER_UPLOAD = int(os.environ['PORTALDS4DS1_MAX_FILES_PER_UPLOAD']) if 'PORTALDS4DS1_CONTRACT_CLIENT_CONTACT' in os.environ: CONTRACT_CLIENT_CONTACT = os.environ['PORTALDS4DS1_CONTRACT_CLIENT_CONTACT'] if 'PORTALDS4DS1_UPLOADS_DIR' in os.environ: UPLOADS_DIR = os.environ['PORTALDS4DS1_UPLOADS_DIR'] if 'PORTALDS4DS1_MAIL_SUBJECT' in os.environ: MAIL_SUBJECT = os.environ['PORTALDS4DS1_MAIL_SUBJECT'] if 'PORTALDS4DS1_MAIL_BODY' in os.environ: MAIL_BODY = os.environ['PORTALDS4DS1_MAIL_BODY'] if 'PORTALDS4DS1_SQL_CONN_STR' in os.environ: SQL_CONN_STR = os.environ['PORTALDS4DS1_SQL_CONN_STR'] ###################### app = Flask(__name__) app.config.update( SECRET_KEY = APP_SECRET_KEY, UPLOADED_PATH = UPLOADS_DIR, MAX_CONTENT_LENGTH = MAX_UPLOAD_SIZE, TEMPLATES_AUTO_RELOAD = True, SQLALCHEMY_DATABASE_URI = SQL_CONN_STR, SQLALCHEMY_ECHO = True ) app.url_map.strict_slashes = False # Run "python app.py db -?" to see more info about DB migrations. manager = Manager(app) db.init_app(app) migrate = Migrate(app, db) manager.add_command('db', MigrateCommand) # Set up dropzone.js to serve all the stuff for "file dropping" on the web interface. dropzone = Dropzone(app) upload_handler_solar = portal.solar.UploadHandlerSolar( UPLOADS_DIR=UPLOADS_DIR, MAIL_HOST=MAIL_HOST, MAIL_LOGIN=MAIL_LOGIN, MAIL_PASS=MAIL_PASS, SMTP_PORT=SMTP_PORT, IMAP_PORT=IMAP_PORT, MAIL_SUBJECT=MAIL_SUBJECT, MAIL_BODY=MAIL_BODY, CONTRACT_CLIENT_CONTACT=CONTRACT_CLIENT_CONTACT, MAX_FILES_PER_UPLOAD=MAX_FILES_PER_UPLOAD, APP_SECRET_KEY=APP_SECRET_KEY ) # Use flask-login to manage user sessions where they are required. login_manager = LoginManager(app) login_manager.init_app(app) def redirect_url(default='/'): return request.args.get('next') or \ request.referrer or \ url_for(default) @app.route('/') def index(): if current_user.is_authenticated: return redirect('/oddaja') return redirect('/login') @login_manager.user_loader def load_user(user_id): user = RegisteredUser.query.get(int(user_id)) return user @app.route('/login') def solar_login_get(): return render_template('solar-login.html') @app.route('/register') def solar_register_get(): return render_template('solar-register.html') @app.route('/login', methods=['POST']) def solar_login_post(): email = request.form.get('email') password = request.form.get('password') remember = True if request.form.get('remember') else False user = portal.solar.get_user_obj_by_email(email) if not user or not check_password_hash(user.pass_hash, password): flash('Napačni podatki za prijavo. Poskusite ponovno.') return redirect('/login') if not user.active: flash('Vaš uporabniški račun še ni bil aktiviran.') return redirect('/login') #portal.solar.add_user_session(user.id) login_user(user, remember=remember) return redirect('/oddaja') @app.route('/register', methods=['POST']) def solar_register_post(): name = request.form.get('name') email = request.form.get('email') password = request.form.get('password') institution_name = request.form.get('institution') institution_role = request.form.get('role') institution = portal.solar.get_institution_obj_by_name(institution_name) user = RegisteredUser.query.filter_by(email=email).first() if user: flash('Uporabniški račun s tem emailom je že registriran.') return redirect('/register') if not name: flash('Prazno polje za ime.') return redirect('/register') if len(name) > 100: flash('Predolgo ime.') return redirect('/register') if not email: flash('Prazno polje za elektronsko pošto.') return redirect('/register') if len(email) > 100: flash('Predolgi email naslov') return redirect('/register') elif not re.search(portal.solar.REGEX_EMAIL, email): flash('Email napačnega formata.') return redirect('/register') if not password: flash('Prazno polje za geslo.') return redirect('/register') if len(password) > 8: flash('Geslo mora biti vsaj 8 znakov dolgo.') return redirect('/register') if len(password) > 100: flash('Predolgo geslo.') return redirect('/register') if institution_role not in ['coordinator', 'mentor', 'other']: flash('Neveljavna vloga v instituciji.') return redirect('/register') if not institution: institution_id = portal.solar.add_institution(institution_name, "") portal.solar.grant_institution_corpus_access(institution_id, "solar") else: institution_id = institution.id user_id = portal.solar.register_new_user(name, email, password, active=False) portal.solar.add_user_to_institution(user_id, institution_id, institution_role) portal.solar.send_admins_new_user_notification_mail(user_id, upload_handler_solar.config) flash('Podatki so bili poslani v potrditev. Ko bo registracija potrjena, boste o tem obveščeni po e-mailu, ki ste ga posredovali zgoraj.') return redirect('/login') @app.route('/logout') @login_required def logout(): logout_user() return redirect('/login') @app.route('/') @login_required def solar(text): is_admin = current_user.role == 'admin' current_user_institution = portal.solar.get_user_institution(current_user.id) current_user_obj = portal.solar.get_user_obj(current_user.get_id()) institution_contract = None if current_user_institution: current_user_institution_coordinator = portal.solar.is_institution_coordinator(current_user.id, current_user_institution.id) institution_contract = portal.solar.get_institution_contract(current_user_institution.id) else: current_user_institution_coordinator = False if text.startswith('oddaja/') or text == 'oddaja': return render_template('solar-oddaja.html', is_admin=is_admin, institution=current_user_institution, institution_contract=institution_contract, is_institution_coordinator=current_user_institution_coordinator) elif text.startswith('zgodovina/') or text == 'zgodovina': upload_items = [] if current_user_institution: upload_items = portal.solar.get_institution_upload_history(current_user_institution.id, n=1000) uploader_names = [] institution_names = [] for item in upload_items: uploader_names.append(portal.solar.get_user_obj(item.upload_user).name) institution = portal.solar.get_institution_obj(item.institution) if not institution: institution_names.append(None) else: institution_names.append(institution.name) return render_template('solar-zgodovina.html', upload_history=upload_items, uploader_names=uploader_names, institution_names=institution_names, is_admin=is_admin, is_institution_coordinator=current_user_institution_coordinator) elif text.startswith('pogodbe-institucije/') or text.startswith('pogodbe-ucencistarsi/'): # Check for ownload contract request. match = re.match('^pogodbe-(institucije|ucencistarsi)/([a-z0-9_]+\.pdf)$', text) if match: contract_type = match.group(1) filename = match.group(2) if len(filename) < 10: return '', 404 prefix = filename[:2] suffix = filename[2:] f_hash = filename.split('.')[0] if contract_type == 'institucije': actual_filename = portal.solar.get_actual_institution_contract_filename(f_hash) else: actual_filename = portal.solar.get_actual_studentparent_contract_filename(f_hash) safe_path = safe_join(str(upload_handler_solar.get_uploads_subdir('contracts')), prefix, suffix) try: return send_file(safe_path, attachment_filename=actual_filename, as_attachment=True) except FileNotFoundError: return '', 404 elif text.startswith('pogodbe/') or text == 'pogodbe': contracts_students = [] contract_school = [] enable_upload_school_contract = False show_upload_form = False collaborators = [] cooperation_history = dict() if current_user_institution: collaborators = portal.solar.get_all_active_institution_users(current_user_institution.id) show_upload_form = True contract_school = portal.solar.get_institution_contract(current_user_institution.id) cooperation_history = portal.solar.get_institution_cooperation_history(current_user_institution.id) if portal.solar.is_institution_coordinator(current_user_obj.id, current_user_institution.id): contracts_students = portal.solar.get_institution_student_contracts(current_user_institution.id) enable_upload_school_contract = True else: contracts_students = portal.solar.get_institution_student_contracts(current_user_institution.id, current_user_obj.id) return render_template('solar-pogodbe.html', contracts_students=contracts_students, contract_school=contract_school, enable_upload_school_contract=enable_upload_school_contract, show_upload_form=show_upload_form, collaborators=collaborators, cooperation_history=cooperation_history, user_id=current_user.id, is_admin=is_admin, is_institution_coordinator=current_user_institution_coordinator) elif text.startswith('admin/') or text == 'admin': users = portal.solar.get_all_users_join_institutions() inactive_users = portal.solar.get_all_users_join_institutions(active=False) solar_institutions = portal.solar.get_all_institutions() cooperation_history = portal.solar.get_cooperation_history() uploads = portal.solar.get_all_upload_history(-1) if is_admin: return render_template('solar-admin.html', users=users, user_cooperation_history=cooperation_history, institutions=solar_institutions, inactive_users=inactive_users, uploads=uploads) elif text.startswith('manage-institution/') or text == 'manage-institution': if portal.solar.is_institution_coordinator(current_user.id, current_user_institution.id): solar_users = portal.solar.get_all_active_users() institution_users = portal.solar.get_all_active_institution_users(current_user_institution.id) return render_template('solar-manage-institution.html', users=solar_users, institution_users=institution_users) return '', 404 @app.route('/pogodbe', methods=['POST']) @login_required def solar_upload_contract(): msg = upload_handler_solar.handle_contract_upload(request, current_user.get_id()) flash(msg) return redirect(redirect_url()) @app.route('/adduser', methods=['POST']) @login_required def solar_add_user(): if not portal.solar.is_admin(current_user.id): return '', 404 name = request.form.get('name') email = request.form.get('email') password = request.form.get('password') if not name: flash('Prazno polje za ime.') return redirect(redirect_url()) if len(name) > 100: flash('Predolgo ime.') return redirect(redirect_url()) if not email: flash('Prazno polje za elektronsko pošto.') return redirect(redirect_url()) if len(email) > 100: flash('Predolg email naslov.') return redirect(redirect_url()) elif not re.search(portal.solar.REGEX_EMAIL, email): flash('Email napačnega formata.') return redirect(redirect_url()) if not password: flash('Prazno polje za geslo.') return redirect(redirect_url()) if len(password) > 100: flash('Predolgo geslo.') return redirect(redirect_url()) user = portal.solar.get_user_obj_by_email(email) if user: #portal.solar.undo_remove_user(user.id) flash('Uporabnik s tem emailom je že vnešen v sistem.') return redirect(redirect_url()) portal.solar.register_new_user(name, email, password) flash('Uporabnik je bil uspešno dodan.') return redirect(redirect_url()) @app.route('/activateuser', methods=['POST']) @login_required def solar_activate_user(): if not portal.solar.is_admin(current_user.id): return '', 404 user_id = request.form.get('id') if not user_id: flash('Prazno polje za ID uporabnika.') return redirect(redirect_url()) rowcount = portal.solar.activate_user(user_id) if rowcount == 0: return '', 404 portal.solar.send_user_activation_mail(user_id, upload_handler_solar.config) flash('Uporabnik je bil aktiviran.') return redirect(redirect_url()) @app.route('/forgotpass') def solar_forgotpass(): return render_template('solar-forgotpass.html') @app.route('/sendresetpass', methods=['POST']) def solar_sendresetpass(): email = request.form.get('email') portal.solar.send_resetpass_mail(email, upload_handler_solar.config) flash('Povezava za ponastavitev gesla je bila poslana na vpisan email naslov.') return redirect(redirect_url()) @app.route('/resetpass/') def solar_resetpass(token): user = portal.solar.verify_reset_token(token, upload_handler_solar.config['APP_SECRET_KEY']) if not user: return '', 404 return render_template('solar-resetpass.html', user=user, token=token) @app.route('/resetpass/', methods=['POST']) def solar_resetpass_post(token): new_password = request.form.get('new_password') user = portal.solar.verify_reset_token(token, upload_handler_solar.config['APP_SECRET_KEY']) if not user: return '', 404 rowcount = portal.solar.update_user_password(user.id, new_password) if rowcount == 0: return '', 404 flash('Ponastavitev gesla je bila uspešna.') return redirect('/login') @app.route('/topuploads') @login_required def solar_topuploads_srednje(): return jsonify(portal.solar.get_top_uploading_institutions()) @app.route('/deluser', methods=['POST']) @login_required def solar_del_user(): if not portal.solar.is_admin(current_user.id): return '', 404 user_id = request.form.get('user_id') portal.solar.remove_user(user_id) flash('Uporabnik je bil odstranjen.') return redirect(redirect_url()) @app.route('/addinstitution', methods=['POST']) @login_required def add_institution(): if not portal.solar.is_admin(current_user.id): return '', 404 name = request.form.get('name') region = request.form.get('region') if not name: flash('Prazno polje za naziv.') return redirect(redirect_url()) if len(name) > 100: flash('Predolgo ime.') return redirect(redirect_url()) if not region in portal.solar.VALID_REGIONS: flash('Neveljavna vrednost za regijo.') return redirect(redirect_url()) institution_id = portal.solar.add_institution(name, region) portal.solar.grant_institution_corpus_access(institution_id, "solar") # TODO: throw out flash('Institucija je bila dodana.') return redirect(redirect_url()) @app.route('/mergeinstitutions', methods=['POST']) @login_required def merge_institutions(): if not portal.solar.is_admin(current_user.id): return '', 404 id_from = request.form.get('id-from') id_to = request.form.get('id-to') if not id_from or not id_to: flash('Prazno polje.') return redirect(redirect_url()) institution_from = portal.solar.get_institution_obj(id_from) institution_to = portal.solar.get_institution_obj(id_to) if not institution_from: flash('Institucija z ID "{}" ne obstaja.'.format(id_from)) return redirect(redirect_url()) if not institution_to: flash('Institucija z ID "{}" ne obstaja.'.format(id_to)) return redirect(redirect_url()) portal.solar.transfer_users_institution(institution_from.id, institution_to.id) portal.solar.transfer_uploads_institution(institution_from.id, institution_to.id) portal.solar.transfer_contracts_institution(institution_from.id, institution_to.id) portal.solar.remove_institution(institution_from.id) flash('Instituciji uspešno združeni') return redirect(redirect_url()) @app.route('/addcooperationhistoryitem', methods=['POST']) @login_required def add_cooperation_history_item(): if not portal.solar.is_admin(current_user.id): return '', 404 user_id = request.form.get('user') institution_id = request.form.get('institution') role = request.form.get('role') school_year = request.form.get('school-year') badge_text = request.form.get('badge-text') user = portal.solar.get_user_obj(user_id) institution = portal.solar.get_institution_obj(institution_id) if not user: flash('Uporabnik s tem ID-jem ne obstaja.') return redirect(redirect_url()) if not institution: flash('Institucija s tem ID-jem ne obstaja.') return redirect(redirect_url()) if not role in ['coordinator', 'mentor', 'other']: flash('Neveljavna vloga "{}".'.format(role)) return redirect(redirect_url()) if not school_year or not re.match('[0-9]{4}/[0-9]{2}', school_year): flash('Šolsko leto mora biti formata "2021/22".') return redirect(redirect_url()) portal.solar.add_cooperation_history_item(user_id, institution_id, role, school_year, badge_text) flash('Vnos dodan.') return redirect(redirect_url()) @app.route('/updateuploaditem', methods=['POST']) @login_required def update_upload_item(): if not portal.solar.is_admin(current_user.id): return '', 404 err_msg = portal.solar.UploadHandlerSolar.check_form(request.form) if err_msg: flash(err_msg) return redirect(redirect_url()) item_id = request.form.get('item-id') program = request.form.get('program') subject = request.form.get('predmet') subject_custom = request.form.get('predmet-custom') grade = request.form.get('letnik') text_type = request.form.get('vrsta') text_type_custom = request.form.get('vrsta-custom') school_year = request.form.get('solsko-leto') grammar_corrections = request.form.get('jezikovni-popravki') rowcount = portal.solar.update_upload_item( item_id, program, subject, subject_custom, grade, text_type, text_type_custom, school_year, grammar_corrections) if rowcount == 0: return '', 404 flash('Vnos spremenjen.') return redirect(redirect_url()) @app.route('/delcooperationhistoryitem', methods=['POST']) @login_required def del_cooperation_history_item(): if not portal.solar.is_admin(current_user.id): return '', 404 entry_id = request.form.get('entry-id') portal.solar.del_cooperation_history_item(entry_id) flash('Vnos odstranjen.') return redirect(redirect_url()) @app.route('/changeinstitutiondata', methods=['POST']) @login_required def change_institution_data(): if not portal.solar.is_admin(current_user.id): return '', 404 institution_id = request.form.get('id') new_name = request.form.get('name') new_region = request.form.get('region') if not new_name: flash('Prazno polje za naziv.') return redirect(redirect_url()) if len(new_name) > 100: flash('Predolgo ime.') return redirect(redirect_url()) if not new_region in portal.solar.VALID_REGIONS: flash('Neveljavna vrednost za regijo.') return redirect(redirect_url()) portal.solar.update_institution_data(institution_id, new_name, new_region) flash('Podatki institucije so bili spremenjeni.') return redirect(redirect_url()) @app.route('/changeuseremail', methods=['POST']) @login_required def change_user_email(): if not portal.solar.is_admin(current_user.id): return '', 404 user_id = request.form.get('user-id') email = request.form.get('email') if not re.search(portal.solar.REGEX_EMAIL, email): flash('Email napačnega formata.') return redirect(redirect_url()) portal.solar.update_user_email(user_id, email) flash('Email spremenjen.') return redirect(redirect_url()) @app.route('/changeuserrole', methods=['POST']) @login_required def change_user_role(): if not portal.solar.is_admin(current_user.id): return '', 404 user_id = request.form.get('user-id') role = request.form.get('role') if not role in ['admin', 'user']: flash('Neveljavna vloga.') return redirect(redirect_url()) portal.solar.update_user_role(user_id, role) flash('Vloga spremenjena.') return redirect(redirect_url()) @app.route('/changeusername', methods=['POST']) @login_required def change_user_name(): if not portal.solar.is_admin(current_user.id): return '', 404 user_id = request.form.get('user-id') name = request.form.get('name') portal.solar.update_user_name(user_id, name) flash('Ime in priimek spremenjena.') return redirect(redirect_url()) @app.route('/addusertoinstitution', methods=['POST']) @login_required def add_user_institution_mapping(): institution_id = request.form.get('institution_id') if not institution_id: institution = portal.solar.get_user_institution(current_user.id) if institution: institution_id = institution.id if not (portal.solar.is_admin(current_user.id) or portal.solar.is_institution_coordinator(current_user.id, institution_id)): return '', 404 user_id = request.form['user_id'] role = request.form['role'] if role not in ['coordinator', 'mentor', 'other']: return '', 404 if portal.solar.get_user_institution(user_id): flash('Uporabnik je že dodeljen instituciji.') return redirect(redirect_url()) portal.solar.add_user_to_institution(user_id, institution_id, role) flash('Uporabnik je bil dodeljen instituciji.') return redirect(redirect_url()) @app.route('/deluserfrominstitution', methods=['POST']) @login_required def del_user_institution_mapping(): user_id = request.form['user_id'] institution = portal.solar.get_user_institution(user_id) if not institution: flash('Uporabnik ni član nobene institucije.') return redirect(redirect_url()) if not portal.solar.is_admin(current_user.id) \ and not portal.solar.is_institution_coordinator(current_user.id, institution.id): flash('Nimate ustreznih pravic za odstranitev uporabnika iz institucije.') return redirect(redirect_url()) portal.solar.del_user_from_institution(user_id, institution.id) flash('Uporabnik je bil odstranjen iz institucije.') return redirect(redirect_url()) @app.route('/upload', methods=['POST']) def handle_upload(): if not current_user.is_authenticated: return '', 404 return upload_handler_solar.handle_upload(request, current_user.get_id()) if __name__ == '__main__': app.run(debug=True)