# Init import os import sys import platform import json import smtplib import zipfile import shutil import logging import random import email.mime.text import email.utils import email.header import urllib.request # Init-Version version_list = json.loads(open('version.json', encoding = 'utf8').read()) print('Version : ' + version_list['beta']['r_ver']) print('DB set version : ' + version_list['beta']['c_ver']) print('Skin set version : ' + version_list['beta']['s_ver']) print('----') # Init-PIP_Install data_up_date = 1 if os.path.exists(os.path.join('data', 'version.json')): data_load_ver = open(os.path.join('data', 'version.json'), encoding = 'utf8').read() if data_load_ver == version_list['beta']['r_ver']: data_up_date = 0 if data_up_date == 1: with open(os.path.join('data', 'version.json'), 'w', encoding = 'utf8') as f: f.write(version_list['beta']['r_ver']) if platform.system() in ('Linux', 'Windows'): os.system( 'python' + ('3' if platform.system() != 'Windows' else '') + ' ' + \ '-m pip install --upgrade --user -r requirements.txt' ) print('----') try: os.execl(sys.executable, sys.executable, *sys.argv) except: pass try: os.execl(sys.executable, '"' + sys.executable + '"', *sys.argv) except: print('Error : restart failed') raise else: print('Error : automatic installation is not supported.') print('Help : try "python3 -m pip install -r requirements.txt"') else: print('PIP check pass') print('----') # Init-Load from .func_mark import * from diff_match_patch import diff_match_patch from gevent.pywsgi import WSGIServer import werkzeug.routing import werkzeug.debug import flask import requests import pymysql import PIL if sys.version_info < (3, 6): import sha3 # Init-Global global_lang = {} global_wiki_set = {} data_css_ver = '106' data_css = '' conn = '' curs = '' # Func # Func-main def load_conn(data): global conn global curs conn = data curs = conn.cursor() load_conn2(data) # Func-init class server_init: def __init__(self): self.env_dict = { 'host' : os.getenv('NAMU_HOST'), 'port' : os.getenv('NAMU_PORT'), 'language' : os.getenv('NAMU_LANG'), 'markup' : os.getenv('NAMU_MARKUP'), 'encode' : os.getenv('NAMU_ENCRYPT') } self.server_set_var = { 'host' : { 'display' : 'Host', 'require' : 'conv', 'default' : '0.0.0.0' }, 'port' : { 'display' : 'Port', 'require' : 'conv', 'default' : '3000' }, 'language' : { 'display' : 'Language', 'require' : 'select', 'default' : 'ko-KR', 'list' : ['ko-KR', 'en-US'] }, 'markup' : { 'display' : 'Markup', 'require' : 'select', 'default' : 'namumark', 'list' : ['namumark', 'custom', 'raw'] }, 'encode' : { 'display' : 'Encryption method', 'require' : 'select', 'default' : 'sha3', 'list' : ['sha3', 'sha256'] } } def server_init(self): return self.server_set_var def init(self, key): if self.env_dict[key] != None: return self.env_dict[key] else: while 1: if self.server_set_var[key]['require'] == 'select': list_ = '[' + ', '.join(self.server_set_var[key]['list']) + ']' else: list_ = '' print('{} ({}) {} : '.format( self.server_set_var[key]['display'], self.server_set_var[key]['default'], list_ ), end = '') server_set_val = input() if server_set_val: if self.server_set_var[key]['require'] == 'select': if server_set_val not in self.server_set_var[key]['list']: pass else: return server_set_val else: return server_set_val else: return self.server_set_var[key]['default'] def get_conn(db_set): if db_set['type'] == 'sqlite': conn = sqlite3.connect(db_set['name'] + '.db') curs = conn.cursor() else: conn = pymysql.connect( host = db_set['mysql_host'], user = db_set['mysql_user'], password = db_set['mysql_pw'], charset = 'utf8mb4', port = int(db_set['mysql_port']) ) curs = conn.cursor() try: curs.execute(db_change( 'create database ' + db_set['name'] + ' ' + \ 'default character set utf8mb4;' )) except: pass conn.select_db(db_set['name']) return conn def update(ver_num, set_data): print('----') # 업데이트 하위 호환 유지 함수 if ver_num < 3160027: print('Add init set') set_init() if ver_num < 3170002: curs.execute(db_change("select html from html_filter where kind = 'extension'")) if not curs.fetchall(): for i in ['jpg', 'jpeg', 'png', 'gif', 'webp']: curs.execute(db_change( "insert into html_filter (html, kind) values (?, 'extension')" ), [i]) if ver_num < 3170400: curs.execute(db_change("select title, sub, code from topic where id = '1'")) for i in curs.fetchall(): curs.execute(db_change( "update topic set code = ? where title = ? and sub = ?" ), [ i[2], i[0], i[1] ]) curs.execute(db_change( "update rd set code = ? where title = ? and sub = ?" ), [ i[2], i[0], i[1] ]) if ver_num < 3171800: curs.execute(db_change("select data from other where name = 'recaptcha'")) change_rec = curs.fetchall() if change_rec and change_rec[0][0] != '': new_rec = re.search(r'data-sitekey="([^"]+)"', change_rec[0][0]) if new_rec: curs.execute(db_change( "update other set data = ? where name = 'recaptcha'" ), [new_rec.group(1)]) else: curs.execute(db_change("update other set data = '' where name = 'recaptcha'")) curs.execute(db_change("update other set data = '' where name = 'sec_re'")) if ver_num < 3172800 and \ set_data['db_type'] == 'mysql': get_data_mysql = json.loads(open('data/mysql.json', encoding = 'utf8').read()) with open('data/mysql.json', 'w') as f: f.write('{ "user" : "' + get_data_mysql['user'] + '", "password" : "' + get_data_mysql['password'] + '", "host" : "localhost" }') if ver_num < 3183603: curs.execute(db_change("select block from ban where band = 'O'")) for i in curs.fetchall(): curs.execute(db_change( "update ban set block = ?, band = 'regex' where block = ? and band = 'O'" ), [ '^' + i[0].replace('.', '\\.'), i[0] ]) curs.execute(db_change("select block from rb where band = 'O'")) for i in curs.fetchall(): curs.execute(db_change( "update rb set block = ?, band = 'regex' where block = ? and band = 'O'" ), [ '^' + i[0].replace('.', '\\.'), i[0] ]) if ver_num < 3190201: today_time = get_time() curs.execute(db_change("select block, end, why, band, login from ban")) for i in curs.fetchall(): curs.execute(db_change( "insert into rb (block, end, today, why, band, login, ongoing) " + \ "values (?, ?, ?, ?, ?, ?, ?)" ), [ i[0], i[1], today_time, i[2], i[3], i[4], '1' ]) if ver_num < 3191301: curs.execute(db_change('' + \ 'select id, title, date from history ' + \ 'where not title like "user:%" ' + \ 'order by date desc ' + \ 'limit 50' + \ '')) data_list = curs.fetchall() for get_data in data_list: curs.execute(db_change( "insert into rc (id, title, date, type) values (?, ?, ?, 'normal')" ), [ get_data[0], get_data[1], get_data[2] ]) if ver_num < 3202400: curs.execute(db_change("select data from other where name = 'update'")) get_data = curs.fetchall() if get_data and get_data[0][0] == 'master': curs.execute(db_change("update other set data = 'beta' where name = 'update'"), []) if ver_num < 3202600: curs.execute(db_change("select name, regex, sub from filter")) for i in curs.fetchall(): curs.execute(db_change( "insert into html_filter (html, kind, plus, plus_t) " + \ "values (?, 'regex_filter', ?, ?)" ), [ i[0], i[1], i[2] ]) curs.execute(db_change("select title, link, icon from inter")) for i in curs.fetchall(): curs.execute(db_change( "insert into html_filter (html, kind, plus, plus_t) " + \ "values (?, 'inter_wiki', ?, ?)"), [ i[0], i[1], i[2] ]) if ver_num < 3203400: curs.execute(db_change("select user, css from custom")) for i in curs.fetchall(): curs.execute(db_change( "insert into user_set (name, id, data) values ('custom_css', ?, ?)" ), [ re.sub(r' \(head\)$', '', i[0]), i[1] ]) if ver_num < 3205500: curs.execute(db_change("select title, decu, dis, view, why from acl")) for i in curs.fetchall(): curs.execute(db_change( "insert into acl (title, data, type) values (?, ?, ?)" ), [i[0], i[1], 'decu']) curs.execute(db_change( "insert into acl (title, data, type) values (?, ?, ?)" ), [i[0], i[2], 'dis']) curs.execute(db_change( "insert into acl (title, data, type) values (?, ?, ?)" ), [i[0], i[3], 'view']) curs.execute(db_change( "insert into acl (title, data, type) values (?, ?, ?)" ), [i[0], i[4], 'why']) if ver_num < 3300101: # 캐시 초기화 curs.execute(db_change('delete from cache_data')) if ver_num < 3300301: # regex_filter 오류 해결 curs.execute(db_change( 'delete from html_filter where kind = "regex_filter" and html is null' )) if ver_num < 3302302: # user이랑 user_set 테이블의 통합 curs.execute(db_change('select id, pw, acl, date, encode from user')) for i in curs.fetchall(): curs.execute(db_change( "insert into user_set (name, id, data) values (?, ?, ?)" ), ['pw', i[0], i[1]]) curs.execute(db_change( "insert into user_set (name, id, data) values (?, ?, ?)" ), ['acl', i[0], i[2]]) curs.execute(db_change( "insert into user_set (name, id, data) values (?, ?, ?)" ), ['date', i[0], i[3]]) curs.execute(db_change( "insert into user_set (name, id, data) values (?, ?, ?)" ), ['encode', i[0], i[4]]) if ver_num < 3400101: # user_set이랑 user_application 테이블의 통합 curs.execute(db_change('' + \ 'select id, pw, date, encode, question, answer, ip, ua, email ' + \ 'from user_application' + \ '')) for i in curs.fetchall(): sql_data = {} sql_data['id'] = i[0] sql_data['pw'] = i[1] sql_data['date'] = i[2] sql_data['encode'] = i[3] sql_data['question'] = i[4] sql_data['answer'] = i[5] sql_data['ip'] = i[6] sql_data['ua'] = i[7] sql_data['email'] = i[8] curs.execute(db_change( "insert into user_set (name, id, data) values (?, ?, ?)" ), ['application', i[0], json.dumps(sql_data)]) conn.commit() print('Update completed') def set_init_always(ver_num): curs.execute(db_change('delete from other where name = "ver"')) curs.execute(db_change('insert into other (name, data) values ("ver", ?)'), [ver_num]) curs.execute(db_change('delete from alist where name = "owner"')) curs.execute(db_change('insert into alist (name, acl) values ("owner", "owner")')) if not os.path.exists(load_image_url()): os.makedirs(load_image_url()) conn.commit() def set_init(): # 초기값 설정 함수 curs.execute(db_change("select html from html_filter where kind = 'email'")) if not curs.fetchall(): for i in ['naver.com', 'gmail.com', 'daum.net', 'kakao.com']: curs.execute(db_change( "insert into html_filter (html, kind) values (?, 'email')" ), [i]) curs.execute(db_change("select html from html_filter where kind = 'extension'")) if not curs.fetchall(): for i in ['jpg', 'jpeg', 'png', 'gif', 'webp']: curs.execute(db_change( "insert into html_filter (html, kind) values (?, 'extension')" ), [i]) curs.execute(db_change( 'select data from other ' + \ 'where name = "smtp_server" or name = "smtp_port" or name = "smtp_security"' )) if not curs.fetchall(): for i in [ ['smtp_server', 'smtp.gmail.com'], ['smtp_port', '587'], ['smtp_security', 'starttls'] ]: curs.execute(db_change( "insert into other (name, data) values (?, ?)" ), [i[0], i[1]]) curs.execute(db_change('select data from other where name = "key"')) rep_data = curs.fetchall() if not rep_data: rep_key = ''.join(random.choice("0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ") for i in range(64)) curs.execute(db_change('insert into other (name, data) values ("key", ?)'), [rep_key]) curs.execute(db_change('select data from other where name = "count_all_title"')) if not curs.fetchall(): curs.execute(db_change('insert into other (name, data) values ("count_all_title", "0")')) conn.commit() # Func-simple def get_default_admin_group(): return ['owner', 'ban'] def load_image_url(): curs.execute(db_change('select data from other where name = "image_where"')) image_where = curs.fetchall() image_where = image_where[0][0] if image_where else 'data/images' return image_where def http_warning(): return '''
''' def load_domain(): curs.execute(db_change("select data from other where name = 'domain'")) domain = curs.fetchall() domain = domain[0][0] if domain and domain[0][0] != '' else flask.request.host_url return domain def load_random_key(long = 64): return ''.join( random.choice( "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" ) for i in range(long) ) def edit_button(): insert_list = [] curs.execute(db_change("select html, plus from html_filter where kind = 'edit_top'")) db_data = curs.fetchall() for get_data in db_data: insert_list += [[get_data[1], get_data[0]]] data = '' for insert_data in insert_list: data += '' + \ '(' + insert_data[1] + ') ' + \ '' data += (' ' if data != '' else '') + '(' + load_lang('add') + ')' data += '
' return data def ip_warning(): if ip_or_user() != 0: curs.execute(db_change('select data from other where name = "no_login_warning"')) data = curs.fetchall() if data and data[0][0] != '': text_data = '' + \ '' + data[0][0] + '' + \ '
' + \ '' else: text_data = '' + \ '' + load_lang('no_login_warning') + '' + \ '
' + \ '' else: text_data = '' return text_data def next_fix(link, num, page, end = 50): list_data = '' if num == 1: if len(page) == end: list_data += '' + \ '
' + \ '(' + load_lang('next') + ')' + \ '' elif len(page) != end: list_data += '' + \ '
' + \ '(' + load_lang('previous') + ')' + \ '' else: list_data += '' + \ '
' + \ '(' + load_lang('previous') + ') (' + load_lang('next') + ')' + \ '' return list_data def leng_check(first, second): if first < second: all_plus = '+' + str(second - first) elif second < first: all_plus = '-' + str(first - second) else: all_plus = '0' return all_plus def number_check(data): try: int(data) return data except: return '1' def redirect(data = '/'): return flask.redirect(flask.request.host_url[:-1] + data) def get_acl_list(type_d = 'normal'): if type_d == 'user': return ['', 'user', 'all'] else: return ['', 'all', 'user', 'admin', 'owner', '50_edit', 'email', 'ban', 'before', '30_day', 'ban_admin'] # Func-login def pw_encode(data, type_d = ''): if type_d == '': curs.execute(db_change('select data from other where name = "encode"')) set_data = curs.fetchall() type_d = set_data[0][0] if type_d == 'sha256': return hashlib.sha256(bytes(data, 'utf-8')).hexdigest() else: if sys.version_info < (3, 6): return sha3.sha3_256(bytes(data, 'utf-8')).hexdigest() else: return hashlib.sha3_256(bytes(data, 'utf-8')).hexdigest() def pw_check(data, data2, type_d = 'no', id_d = ''): curs.execute(db_change('select data from other where name = "encode"')) db_data = curs.fetchall() if type_d != 'no': if type_d == '': set_data = 'sha3' else: set_data = type_d else: set_data = db_data[0][0] re_data = 1 if pw_encode(data, set_data) == data2 else 0 if db_data[0][0] != set_data and re_data == 1 and id_d != '': curs.execute(db_change("update user_set set data = ? where id = ? and name = 'pw'"), [ pw_encode(data), id_d ]) curs.execute(db_change("update user_set set data = ? where id = ? and name = 'encode'"), [ db_data[0][0], id_d ]) return re_data # Func-skin def easy_minify(data, tool = None): return data def load_lang(data, safe = 0): global global_lang ip = ip_check() if ip_or_user(ip) == 0: curs.execute(db_change('select data from user_set where name = "lang" and id = ?'), [ip]) rep_data = curs.fetchall() elif 'lang' in flask.session: rep_data = [[flask.session['lang']]] else: curs.execute(db_change("select data from other where name = 'language'")) rep_data = curs.fetchall() if not rep_data or rep_data[0][0] in ('', 'default'): curs.execute(db_change("select data from other where name = 'language'")) rep_data = curs.fetchall() if rep_data: lang_name = rep_data[0][0] else: lang_name = 'en-US' if lang_name in global_lang: lang = global_lang[lang_name] else: lang_list = os.listdir('lang') if (lang_name + '.json') in lang_list: lang = json.loads(open( os.path.join('lang', lang_name + '.json'), encoding = 'utf8' ).read()) global_lang[lang_name] = lang else: lang = {} if data in lang: if safe == 1: return lang[data] else: return html.escape(lang[data]) return html.escape(data + ' (' + lang_name + ')') def skin_check(set_n = 0): # 개편 필요? skin_list = load_skin('tenshi', 1) skin = skin_list[0] ip = ip_check() user_need_skin = '' if ip_or_user(ip) == 0: curs.execute(db_change('select data from user_set where name = "skin" and id = ?'), [ip]) skin_exist = curs.fetchall() if skin_exist: user_need_skin = skin_exist[0][0] else: if 'skin' in flask.session: user_need_skin = flask.session['skin'] if user_need_skin == '': curs.execute(db_change('select data from other where name = "skin"')) skin_exist = curs.fetchall() if skin_exist: user_need_skin = skin_exist[0][0] if user_need_skin != '' and user_need_skin in skin_list: skin = user_need_skin if set_n == 0: return './views/' + skin + '/index.html' else: return skin def wiki_css(data): global data_css global data_css_ver data += ['' for _ in range(0, 3 - len(data))] if data_css == '': for i_data in os.listdir(os.path.join("views", "main_css", "css")): if i_data != 'sub': data_css += '' for i_data in os.listdir(os.path.join("views", "main_css", "js")): if i_data != 'sub': data_css += '' data = data[0:2] + ['', ''' ''' + data_css + ''] + data[2:] return data def cut_100(data): if re.search(r'^\/w\/', flask.request.path): data = re.sub(r'' + \ '
' + \ '
' + \ '' else: data += '' + \ '' + \ '' + \ '' + \ '' return data def captcha_post(re_data, num = 1): if num == 1: curs.execute(db_change('select data from other where name = "sec_re"')) sec_re = curs.fetchall() if sec_re and sec_re[0][0] != '' and \ ip_or_user() != 0 and captcha_get() != '': data = requests.get( 'https://www.google.com/recaptcha/api/siteverify' + \ '?secret=' + sec_re[0][0] + '&response=' + re_data ) if data.status_code == 200: json_data = json.loads(data.text) if json_data['success'] != True: return 1 return 0 # Func-user def ip_or_user(data = ''): # 1 == ip # 0 == reg if data == '': data = ip_check() if re.search(r'(\.|:)', data): return 1 else: return 0 def admin_check(num = None, what = None, name = ''): ip = ip_check() if name == '' else name time_data = get_time() pass_ok = 0 if ip_or_user(ip) == 0: curs.execute(db_change( "select data from user_set where id = ? and name = 'acl'" ), [ip]) user_auth = curs.fetchall() if user_auth: user_auth = user_auth[0][0] check = { 0 : 'owner', 1 : 'ban', 2 : 'nothing', 3 : 'toron', 4 : 'check', 5 : 'acl', 6 : 'hidel', 7 : 'give' } if not num: check = check[0] elif num == 'all': check = [check[i] for i in check] else: check = check[num] curs.execute(db_change( 'select name from alist where name = ? and acl = "owner"' ), [user_auth]) if curs.fetchall(): pass_ok = 1 else: if num == 'all': curs.execute(db_change( 'select name from alist where name = ?' ), [user_auth]) else: curs.execute(db_change( 'select name from alist where name = ? and acl = ?' ), [user_auth, check]) if curs.fetchall(): pass_ok = 1 if pass_ok == 1: if what: curs.execute(db_change( "insert into re_admin (who, what, time) values (?, ?, ?)" ), [ip, what, time_data]) conn.commit() return 1 return 0 def acl_check(name = 'test', tool = '', topic_num = '1'): ip = ip_check() get_ban = ban_check() acl_c = re.search(r"^user:((?:(?!\/).)*)", name) if name else None if tool == '' and acl_c: acl_n = acl_c.groups() if get_ban == 1: return 1 if admin_check(5) == 1: return 0 curs.execute(db_change( "select data from acl where title = ? and type = 'decu'" ), ['user:' + acl_n[0]]) acl_data = curs.fetchall() if acl_data: if acl_data[0][0] == 'all': return 0 elif acl_data[0][0] == 'user' and not ip_or_user(ip) == 1: return 0 elif ip == acl_n[0] and not ip_or_user(ip) == 1: return 0 else: if ip == acl_n[0] and not ip_or_user(ip) == 1 and not ip_or_user(acl_n[0]) == 1: return 0 return 1 if tool == 'topic': if not name: curs.execute(db_change("select title from rd where code = ?"), [topic_num]) name = curs.fetchall() name = name[0][0] if name else 'test' end = 3 elif tool == 'render' or tool == '' or tool == 'vote': if tool == '' and acl_check(name, 'render') == 1: return 1 end = 2 else: end = 1 for i in range(0, end): if tool == '': if i == 0: curs.execute(db_change( "select data from acl where title = ? and type = 'decu'" ), [name]) else: curs.execute(db_change('select data from other where name = "edit"')) num = 5 elif tool == 'topic': if i == 0 and topic_num: curs.execute(db_change("select acl from rd where code = ?"), [topic_num]) elif i == 1: curs.execute(db_change( "select data from acl where title = ? and type = 'dis'" ), [name]) else: curs.execute(db_change('select data from other where name = "discussion"')) num = 3 elif tool == 'upload': curs.execute(db_change("select data from other where name = 'upload_acl'")) num = 5 elif tool == 'many_upload': curs.execute(db_change("select data from other where name = 'many_upload_acl'")) num = 5 elif tool == 'vote': if i == 0: curs.execute(db_change( 'select acl from vote where id = ? and user = ""' ), [topic_num]) else: curs.execute(db_change('select data from other where name = "vote_acl"')) num = None else: # tool == 'render' if i == 0: curs.execute(db_change( "select data from acl where title = ? and type = 'view'" ), [name]) else: curs.execute(db_change("select data from other where name = 'all_view_acl'")) num = 5 acl_data = curs.fetchall() if ( i == (end - 1) and \ (not acl_data or acl_data[0][0] == '' or acl_data[0][0] == 'normal') ) and \ get_ban == 1 and \ tool != 'render': return 1 elif acl_data and acl_data[0][0] != 'normal' and acl_data[0][0] != '': if acl_data[0][0] != 'ban' and get_ban == 1 and tool != 'render': return 1 if acl_data[0][0] == 'all' or acl_data[0][0] == 'ban': return 0 elif acl_data[0][0] == 'user': if ip_or_user(ip) != 1: return 0 elif acl_data[0][0] == 'admin': if ip_or_user(ip) != 1: if admin_check(num) == 1: return 0 elif acl_data[0][0] == '50_edit': if ip_or_user(ip) != 1: if admin_check(num) == 1: return 0 else: curs.execute(db_change( "select count(*) from history where ip = ?" ), [ip]) count = curs.fetchall() count = count[0][0] if count else 0 if count >= 50: return 0 elif acl_data[0][0] == 'before': if ip_or_user(ip) != 1: if admin_check(num) == 1: return 0 curs.execute(db_change( "select ip from history where title = ? and ip = ?" ), [name, ip]) if curs.fetchall(): return 0 elif acl_data[0][0] == '30_day': if ip_or_user(ip) != 1: if admin_check(num) == 1: return 0 else: curs.execute(db_change( "select data from user_set where id = ? and name = 'date'" ), [ip]) user_date = curs.fetchall()[0][0] time_1 = datetime.datetime.strptime( user_date, '%Y-%m-%d %H:%M:%S' ) + datetime.timedelta(days = 30) time_2 = datetime.datetime.strptime( get_time(), '%Y-%m-%d %H:%M:%S' ) if time_2 > time_1: return 0 elif acl_data[0][0] == 'email': if ip_or_user(ip) != 1: if admin_check(num) == 1: return 0 else: curs.execute(db_change( "select data from user_set where id = ? and name = 'email'" ), [ip]) if curs.fetchall(): return 0 elif acl_data[0][0] == 'owner': if admin_check() == 1: return 0 elif acl_data[0][0] == 'ban_admin': if admin_check(1) == 1 or ban_check() == 1: return 0 return 1 else: if i == (end - 1): if tool == 'topic' and topic_num: curs.execute(db_change( "select title from rd where code = ? and stop != ''" ), [topic_num]) if curs.fetchall(): if admin_check(3, 'topic (code ' + topic_num + ')') == 1: return 0 else: return 0 else: return 0 return 1 def ban_check(ip = None, tool = ''): ip = ip_check() if not ip else ip tool = '' if not tool else tool if admin_check(None, None, ip) == 1: return 0 curs.execute(db_change( "update rb set ongoing = '' " + \ "where end < ? and end != '' and ongoing = '1'" ), [get_time()]) conn.commit() curs.execute(db_change("" + \ "select login, block from rb " + \ "where band = 'regex' and ongoing = '1'" + \ "")) regex_d = curs.fetchall() for test_r in regex_d: g_regex = re.compile(test_r[1]) if g_regex.search(ip): if tool == 'login': if test_r[0] != 'O': return 1 else: return 1 curs.execute(db_change("" + \ "select login from rb " + \ "where block = ? and band = '' and ongoing = '1'" + \ "" + \ ""), [ip]) ban_d = curs.fetchall() if ban_d: if tool == 'login': if ban_d[0][0] != 'O': return 1 else: return 1 return 0 def ip_pas(raw_ip, type_d = 0): hide = 0 end_ip = {} i = 0 return_data = 0 if type(raw_ip) != type([]): get_ip = [raw_ip] return_data = 1 else: get_ip = raw_ip curs.execute(db_change("select data from other where name = 'ip_view'")) ip_view = curs.fetchall() ip_view = ip_view[0][0] if ip_view else '' ip_view = '' if admin_check(1) == 1 else ip_view get_ip = list(set(get_ip)) for raw_ip in get_ip: change_ip = 0 is_this_ip = ip_or_user(raw_ip) if is_this_ip != 0 and ip_view != '': ip = re.sub(r'\.([^.]*)\.([^.]*)$', '.*.*', raw_ip) ip = re.sub(r':([^:]*):([^:]*)$', ':*:*', ip) change_ip = 1 else: ip = raw_ip if type_d == 0: if is_this_ip == 0: ip = '' + raw_ip + '' if change_ip == 0: ip += ' (' + load_lang('tool') + ')' end_ip[raw_ip] = ip if return_data == 1: return end_ip[raw_ip] else: return end_ip # Func-edit def slow_edit_check(): curs.execute(db_change("select data from other where name = 'slow_edit'")) slow_edit = curs.fetchall() if slow_edit and slow_edit != '0' and admin_check(5) != 1: slow_edit = slow_edit[0][0] curs.execute(db_change( "select date from history where ip = ? order by date desc limit 1" ), [ip_check()]) last_edit_data = curs.fetchall() if last_edit_data: last_edit_data = int(re.sub(' |:|-', '', last_edit_data[0][0])) now_edit_data = int( (datetime.datetime.now() - datetime.timedelta( seconds = int(slow_edit)) ).strftime("%Y%m%d%H%M%S") ) if last_edit_data > now_edit_data: return 1 return 0 def edit_filter_do(data): if admin_check(1) != 1: curs.execute(db_change( "select plus, plus_t from html_filter where kind = 'regex_filter' and plus != ''" )) for data_list in curs.fetchall(): match = re.compile(data_list[0], re.I) if match.search(data): ban_insert( ip_check(), '0' if data_list[1] == 'X' else data_list[1], 'edit filter', None, 'tool:edit filter' ) return 1 return 0 # Func-insert def add_alarm(who, context): curs.execute(db_change( 'insert into alarm (name, data, date) values (?, ?, ?)' ), [who, context, get_time()]) def add_user(user_name, user_pw, user_email = '', user_encode = ''): if user_encode == '': user_pw_hash = pw_encode(user_pw) curs.execute(db_change('select data from other where name = "encode"')) data_encode = curs.fetchall() data_encode = data_encode[0][0] else: user_pw_hash = user_pw data_encode = user_encode curs.execute(db_change("select id from user_set limit 1")) if not curs.fetchall(): user_auth = 'owner' else: user_auth = 'user' curs.execute(db_change("insert into user_set (id, name, data) values (?, 'pw', ?)"), [ user_name, user_pw_hash ]) curs.execute(db_change("insert into user_set (id, name, data) values (?, 'acl', ?)"), [ user_name, user_auth ]) curs.execute(db_change("insert into user_set (id, name, data) values (?, 'date', ?)"), [ user_name, get_time() ]) curs.execute(db_change("insert into user_set (id, name, data) values (?, 'encode', ?)"), [ user_name, data_encode ]) if user_email != '': curs.execute(db_change("insert into user_set (name, id, data) values ('email', ?, ?)"), [ user_name, user_email ]) conn.commit() def ua_plus(u_id, u_ip, u_agent, time): curs.execute(db_change("select data from other where name = 'ua_get'")) rep_data = curs.fetchall() if rep_data and rep_data[0][0] != '': pass else: curs.execute(db_change( "insert into ua_d (name, ip, ua, today, sub) values (?, ?, ?, ?, '')" ), [ u_id, u_ip, u_agent, time ]) def ban_insert(name, end, why, login, blocker, type_d = None): now_time = get_time() band = type_d if type_d else '' curs.execute(db_change( "update rb set ongoing = '' where end < ? and end != '' and ongoing = '1'" ), [now_time]) curs.execute(db_change("" + \ "select block from rb " + \ "where ((end > ? and end != '') or end = '') and block = ? and " + \ "band = ? and ongoing = '1'" + \ ""), [now_time, name, band]) if curs.fetchall(): curs.execute(db_change( "insert into rb (block, end, today, blocker, why, band) values (?, ?, ?, ?, ?, ?)" ), [ name, 'release', now_time, blocker, '', band ]) curs.execute(db_change( "update rb set ongoing = '' where block = ? and band = ? and ongoing = '1'" ), [name, band]) else: login = 'O' if login != '' else '' if end != '0': end = int(number_check(end)) time = datetime.datetime.now() plus = datetime.timedelta(seconds = end) r_time = (time + plus).strftime("%Y-%m-%d %H:%M:%S") else: r_time = '' curs.execute(db_change( "insert into rb (block, end, today, blocker, why, band, ongoing, login) " + \ "values (?, ?, ?, ?, ?, ?, '1', ?)" ), [ name, r_time, now_time, blocker, why, band, login ]) conn.commit() def rd_plus(topic_num, date, name = None, sub = None): curs.execute(db_change("select code from rd where code = ?"), [topic_num]) if curs.fetchall(): curs.execute(db_change("update rd set date = ? where code = ?"), [date, topic_num]) else: curs.execute(db_change( "insert into rd (title, sub, code, date) values (?, ?, ?, ?)" ), [name, sub, topic_num, date]) conn.commit() def history_plus(title, data, date, ip, send, leng, t_check = '', mode = ''): if mode == 'add': curs.execute(db_change( "select id from history where title = ? order by id + 0 asc limit 1" ), [title]) id_data = curs.fetchall() id_data = str(int(id_data[0][0]) - 1) if id_data else '0' else: curs.execute(db_change( "select id from history where title = ? order by id + 0 desc limit 1" ), [title]) id_data = curs.fetchall() id_data = str(int(id_data[0][0]) + 1) if id_data else '1' mode = mode if not re.search('^user:', title) else 'user' send = re.sub(r'\(|\)|<|>', '', send) send = send[:128] if len(send) > 128 else send send = send + ' (' + t_check + ')' if t_check != '' else send if mode != 'add' and mode != 'user': curs.execute(db_change("select count(*) from rc where type = 'normal'")) if curs.fetchall()[0][0] >= 200: curs.execute(db_change( "select id, title from rc where type = 'normal' order by date asc limit 1" )) rc_data = curs.fetchall() if rc_data: curs.execute(db_change( 'delete from rc where id = ? and title = ? and type = "normal"' ), [ rc_data[0][0], rc_data[0][1] ]) curs.execute(db_change( "insert into rc (id, title, date, type) values (?, ?, ?, 'normal')" ), [ id_data, title, date ]) if mode != 'add': curs.execute(db_change("select count(*) from rc where type = ?"), [mode]) if curs.fetchall()[0][0] >= 200: curs.execute(db_change( "select id, title from rc where type = ? order by date asc limit 1" ), [mode]) rc_data = curs.fetchall() if rc_data: curs.execute(db_change( 'delete from rc where id = ? and title = ? and type = ?' ), [ rc_data[0][0], rc_data[0][1], mode ]) curs.execute(db_change( "insert into rc (id, title, date, type) values (?, ?, ?, ?)" ), [ id_data, title, date, mode ]) curs.execute(db_change( "insert into history (id, title, data, date, ip, send, leng, hide, type) " + \ "values (?, ?, ?, ?, ?, ?, ?, '', ?)" ), [ id_data, title, data, date, ip, send, leng, mode ]) # Func-error def re_error(data): conn.commit() if data == '/ban': if ban_check() == 1: end = '
' else: end = '' return easy_minify(flask.render_template(skin_check(), imp = [load_lang('error'), wiki_set(1), wiki_custom(), wiki_css([0, 0])], data = '

' + load_lang('error') + '

' + end, menu = 0 )), 401 else: num = int(number_check(data.replace('/error/', ''))) if num == 1: data = load_lang('no_login_error') elif num == 2: data = load_lang('no_exist_user_error') elif num == 3: data = load_lang('authority_error') elif num == 4: data = load_lang('no_admin_block_error') elif num == 5: data = load_lang('skin_error') elif num == 6: data = load_lang('same_id_exist_error') elif num == 7: data = load_lang('long_id_error') elif num == 8: data = load_lang('id_char_error') + ' (' + load_lang('id_filter_list') + ')' elif num == 9: data = load_lang('file_exist_error') elif num == 10: data = load_lang('password_error') elif num == 11: data = load_lang('topic_long_error') elif num == 12: data = load_lang('email_error') elif num == 13: data = load_lang('recaptcha_error') elif num == 14: data = load_lang('file_extension_error') + ' (' + load_lang('extension_filter_list') + ')' elif num == 15: data = load_lang('edit_record_error') elif num == 16: data = load_lang('same_file_error') elif num == 17: data = load_lang('file_capacity_error') + wiki_set(3) elif num == 18: data = load_lang('email_send_error') elif num == 19: data = load_lang('decument_exist_error') elif num == 20: data = load_lang('password_diffrent_error') elif num == 21: data = load_lang('edit_filter_error') elif num == 22: data = load_lang('file_name_error') elif num == 23: data = load_lang('regex_error') elif num == 24: curs.execute(db_change("select data from other where name = 'slow_edit'")) slow_edit = curs.fetchall() slow_edit = '' if not slow_edit else slow_edit[0][0] data = load_lang('fast_edit_error') + slow_edit elif num == 25: data = load_lang('too_many_dec_error') elif num == 26: data = load_lang('application_not_found') elif num == 27: data = load_lang("invalid_password_error") elif num == 28: data = load_lang('watchlist_overflow_error') elif num == 29: data = load_lang('copyright_disagreed') elif num == 30: data = load_lang('ie_wrong_callback') elif num == 33: data = load_lang('restart_fail_error') elif num == 34: data = load_lang("update_error") + ' (Github)' elif num == 35: data = load_lang('same_email_error') elif num == 36: data = load_lang('input_email_error') else: data = '???' if num == 5: if flask.request.path != '/main_skin_set': title = load_lang('skin_set') tool = [['main_skin_set', load_lang('main_skin_set')]] load_skin_set = '' else: title = load_lang('main_skin_set') tool = [['skin_set', load_lang('skin_set')]] load_skin_set = '' return easy_minify(flask.render_template(skin_check(), imp = [title, wiki_set(1), wiki_custom(), wiki_css([0, 0])], data = '' + \ '
' + \ '

' + load_lang('error') + '

' + \ '' + \ '
' + \ load_skin_set, menu = tool )) else: return easy_minify(flask.render_template(skin_check(), imp = [load_lang('error'), wiki_set(1), wiki_custom(), wiki_css([0, 0])], data = '' + \ '

' + load_lang('error') + '

' + \ '' + \ '', menu = 0 )), 400