\n\n' if not sort.startswith('-'): sort = '+%s' % sort if sort[1:] == 'time': urls = sorted(urlserver['urls'].items()) else: idx = ['time', 'nick', 'buffer'].index(sort[1:]) urls = sorted(urlserver['urls'].items(), key=lambda url: url[1][idx].lower()) if sort.startswith('-'): urls.reverse() sortkey = { '-': ('', '↑'), '+': ('-', '↓') } content += ' ' for column, defaultsort in (('time', '-'), ('nick', ''), ('buffer', '')): if sort[1:] == column: content += ('' % ( column, sortkey[sort[0]][0], column, column.capitalize(), sortkey[sort[0]][1])) else: content += ('' % ( column, defaultsort, column, column.capitalize())) content += '' content += '\n' for key, item in urls: content += ' ' url = item[3] obj = '' message = (html.escape(item[4].replace(url, '\x01\x02\x03\x04')) .split('\t', 1)) message[0] = '%s' % message[0] message[1] = '%s' % message[1] strjoin = (' %s ' % urlserver_settings['http_prefix_suffix'] .replace(' ', ' ')) target = '' if urlserver_settings['http_open_in_new_page'] == 'on': target = ' target=_blank' message = strjoin.join(message).replace( '\x01\x02\x03\x04', '%s' '' % ( urlserver_short_url(key, False), url, target, url)) if urlserver_settings['http_embed_image'] == 'on' and \ url.lower().endswith(('.jpg', '.jpeg', '.png', '.gif', '.bmp', '.svg')): obj = ('

' '

' % (url, url, url)) elif urlserver_settings['http_embed_youtube'] == 'on' and \ 'youtube.com/' in url: m = re.search('v=([\w\d]+)', url) if m: yid = m.group(1) try: size = (urlserver_settings['http_embed_youtube_size'] .split('*')) width = int(size[0]) height = int(size[1]) except: width = 480 height = 350 obj = ('

' '

' % (yid, width, height, yid)) content += ('' '' '\n' % (message, obj) content += '

' '%s %s	' '%s	URLs
%s	%s	%s	' % ( item[0], item[1], item[2])) content += '%s%s

' if len(urlserver_settings['http_css_url']) > 0: css = ('' % urlserver_settings['http_css_url']) else: css = ('\n' % ( urlserver_settings['http_bg_color'], urlserver_settings['http_fg_color'])) html_code = ('\n' '\n' '%s\n' '\n' '%s\n' '\n' '\n' '\n' '\n%s\n\n' '' % ( urlserver_settings['http_title'], css, urlserver_get_base_url(), content)) urlserver_server_reply(conn, '200 OK', '', html_code) def urlserver_check_auth(data): """Check user/password to access a page/URL.""" global urlserver_settings if not urlserver_settings['http_auth']: return True http_auth = weechat.string_eval_expression( urlserver_settings['http_auth'], {}, {}, {}) auth = re.search('^Authorization: Basic (\S+)$', data, re.MULTILINE | re.IGNORECASE) if auth and (base64_decode(auth.group(1)).decode('utf-8') == http_auth): return True return False def urlserver_server_fd_cb(data, fd): """Callback for server socket.""" global urlserver, urlserver_settings if not urlserver['socket']: return weechat.WEECHAT_RC_OK conn, addr = urlserver['socket'].accept() if urlserver_settings['debug'] == 'on': weechat.prnt('', 'urlserver: connection from %s' % str(addr)) if urlserver_settings['http_allowed_ips'] and \ not re.match(urlserver_settings['http_allowed_ips'], addr[0]): if urlserver_settings['debug'] == 'on': weechat.prnt('', 'urlserver: IP not allowed') conn.close() return weechat.WEECHAT_RC_OK data = None try: conn.settimeout(0.3) data = conn.recv(4096).decode('utf-8') data = data.replace('\r\n', '\n') except: return weechat.WEECHAT_RC_OK replysent = False sort = '-time' referer = re.search('^Referer:', data, re.MULTILINE | re.IGNORECASE) m = re.search('^GET /(.*) HTTP/.*$', data, re.MULTILINE) if m: url = m.group(1) if urlserver_settings['debug'] == 'on': weechat.prnt('', 'urlserver: %s' % m.group(0)) if 'favicon.' in url: urlserver_server_reply(conn, '200 OK', '', urlserver_server_favicon(), mimetype='image/x-icon') replysent = True else: # check if prefix is ok (if prefix defined in settings) prefixok = True if urlserver_settings['http_url_prefix']: if url.startswith(urlserver_settings['http_url_prefix']): url = url[len(urlserver_settings['http_url_prefix']):] if url.startswith('/'): url = url[1:] else: prefixok = False # prefix ok, go on with url if prefixok: if url.startswith('sort='): # sort asked for list of urls sort = url[5:] url = '' if url: # short url, read base62 key and redirect to page number = -1 try: number = base62_decode(url) except: pass if number >= 0 and number in urlserver['urls']: authok = ( urlserver_settings['http_auth_redirect'] != 'on' or urlserver_check_auth(data) ) if authok: # if we have a referer in request, use meta for # redirection (so that referer is not sent) # otherwise, we can make redirection with HTTP 302 if referer: urlserver_server_reply( conn, '200 OK', '', '\n' '' % urlserver['urls'][number][3]) else: conn.sendall( 'HTTP/1.1 302\r\n' 'Location: {}\r\n\r\n' .format(urlserver['urls'][number][3]) .encode('utf-8')) else: urlserver_server_reply_auth_required(conn) replysent = True else: # page with list of urls if urlserver_check_auth(data): urlserver_server_reply_list(conn, sort) else: urlserver_server_reply_auth_required(conn) replysent = True else: if urlserver_settings['debug'] == 'on': weechat.prnt('', 'urlserver: prefix missing') if not replysent: urlserver_server_reply(conn, '404 Not found', '', '\n' 'Page not found\n' '

Page not found

\n' '') conn.close() return weechat.WEECHAT_RC_OK def urlserver_server_status(): """Display status of server.""" global urlserver if urlserver['socket']: weechat.prnt('', 'URL server listening on %s' % str(urlserver['socket'].getsockname())) else: weechat.prnt('', 'URL server not running') def urlserver_server_start(): """Start mini HTTP server.""" global urlserver, urlserver_settings if urlserver['socket']: weechat.prnt('', 'URL server already running') return port = 0 try: port = int(urlserver_settings['http_port']) except: port = 0 urlserver['socket'] = socket.socket(socket.AF_INET, socket.SOCK_STREAM) urlserver['socket'].setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: urlserver['socket'].bind((urlserver_settings['http_hostname'] or socket.getfqdn(), port)) except Exception as e: weechat.prnt('', '%sBind error: %s' % (weechat.prefix('error'), e)) urlserver['socket'] = None urlserver_server_status() return urlserver['socket'].listen(5) urlserver['hook_fd'] = weechat.hook_fd(urlserver['socket'].fileno(), 1, 0, 0, 'urlserver_server_fd_cb', '') urlserver_server_status() def urlserver_server_stop(): """Stop mini HTTP server.""" global urlserver if urlserver['socket'] or urlserver['hook_fd']: if urlserver['socket']: urlserver['socket'].close() urlserver['socket'] = None if urlserver['hook_fd']: weechat.unhook(urlserver['hook_fd']) urlserver['hook_fd'] = None weechat.prnt('', 'URL server stopped') def urlserver_server_restart(): """Restart mini HTTP server.""" urlserver_server_stop() urlserver_server_start() def urlserver_display_url_detail(key, return_url=False): global urlserver url = urlserver['urls'][key] nick = url[1] if nick: nick += ' @ ' if return_url: return urlserver_short_url(key) else: weechat.prnt_date_tags( urlserver['buffer'], 0, 'notify_none', '%s, %s%s%s%s: %s%s%s -> %s' % ( url[0], nick, weechat.color('chat_buffer'), url[2], weechat.color('reset'), weechat.color(urlserver_settings['color']), urlserver_short_url(key), weechat.color('reset'), url[3])) def urlserver_buffer_input_cb(data, buffer, input_data): if input_data in ('q', 'Q'): weechat.buffer_close(buffer) return weechat.WEECHAT_RC_OK def urlserver_buffer_close_cb(data, buffer): global urlserver urlserver['buffer'] = '' return weechat.WEECHAT_RC_OK def urlserver_open_buffer(): global urlserver, urlserver_settings if not urlserver['buffer']: urlserver['buffer'] = weechat.buffer_new( SCRIPT_BUFFER, 'urlserver_buffer_input_cb', '', 'urlserver_buffer_close_cb', '') if urlserver['buffer']: weechat.buffer_set(urlserver['buffer'], 'title', 'urlserver') weechat.buffer_set(urlserver['buffer'], 'localvar_set_no_log', '1') weechat.buffer_set(urlserver['buffer'], 'time_for_each_line', '0') weechat.buffer_set(urlserver['buffer'], 'print_hooks_enabled', '0') weechat.buffer_clear(urlserver['buffer']) keys = sorted(urlserver['urls']) for key in keys: urlserver_display_url_detail(key) weechat.buffer_set(urlserver['buffer'], 'display', '1') def urlserver_cmd_cb(data, buffer, args): """The /urlserver command.""" global urlserver if args == 'start': urlserver_server_start() elif args == 'restart': urlserver_server_restart() elif args == 'stop': urlserver_server_stop() elif args == 'status': urlserver_server_status() elif args == 'clear': urlserver['urls'] = {} urlserver['number'] = 0 weechat.prnt('', 'urlserver: list cleared') else: urlserver_open_buffer() return weechat.WEECHAT_RC_OK def urlserver_update_urllist(buffer_full_name, buffer_short_name, tags, prefix, message, nick=None): """Update urls list and return a list of short urls for message.""" global urlserver, urlserver_settings # skip ignored buffers if urlserver_settings['msg_ignore_buffers'] and \ buffer_full_name in (urlserver_settings['msg_ignore_buffers'] .split(',')): return None listtags = [] if tags: listtags = tags.split(',') # skip ignored tags if urlserver_settings['msg_ignore_tags']: for itag in urlserver_settings['msg_ignore_tags'].split(','): for tag in listtags: if tag.startswith(itag): return None # exit if a required tag is missing if urlserver_settings['msg_require_tags']: for rtag in urlserver_settings['msg_require_tags'].split(','): tagfound = False for tag in listtags: if tag.startswith(rtag): tagfound = True break if not tagfound: return None # ignore message is matching the "msg_ignore_regex" if urlserver_settings['msg_ignore_regex']: if re.search(urlserver_settings['msg_ignore_regex'], prefix + '\t' + message): return None # extract nick from tags if not nick: nick = '' for tag in listtags: if tag.startswith('nick_'): nick = tag[5:] break # get URL min length min_length = 0 try: min_length = int(urlserver_settings['url_min_length']) # Detect the minimum length based on shorten url length if min_length == -1: min_length = len(urlserver_short_url(urlserver['number'])) + 1 except: min_length = 0 # shorten URL(s) in message urls_short = [] for match in urlserver['regex'].finditer(message): url = match.group('url') trailer = match.group('trailer') # Heuristics for dealing with valid URI characters used as URI # delimiters. if url[-1] == ',': # Does the URL contain other commas? If so, don't strip. # Is the URL followed by a space? If not, don't strip. if trailer == ' ' and url[:-1].count(',') == 0: url = url[:-1] elif url[-1] == '.': # Strip if the URL is followed by whitespace *or* nothing. # Nothing seems to use a . at the end, and it's a natural # sentence terminator. if trailer is None or trailer == ' ': url = url[:-1] elif url[-1] == ')' or url[-1] == ']': # Tough one. First check whether the URL is followed by # a space or end of line. if trailer is None or trailer == ' ': closer = url[-1] if closer == ')': opener = '(' elif closer == ']': opener = '[' # Check if the brackets would be balanced inside the URL. opening = url.count(opener) closing = url.count(closer) if opening < closing: match_start, match_end = match.span('url') # Is the URL *immediately* preceded by an opener? prior = message[:match_start] if prior and prior[-1] == opener: url = url[:-1] else: after = message[match_end:] # Are brackets outside of the URL unbalanced? opening = prior.count(opener) + after.count(opener) closing = prior.count(closer) + after.count(closer) if opening > closing: url = url[:-1] elif url[-1] == "'": # Another doozy. Can't really work with balance because of # contractions such as "can't". # So let's simply check for enclosing, but only if there's not # another delimiter. if trailer is None or trailer == ' ': match_start = match.start('url') if match_start > 0 and message[match_start - 1] == "'": url = url[:-1] # End heuristics if len(url) >= min_length: if urlserver_settings['msg_ignore_dup_urls'] == 'on': same_urls = [key for key, value in urlserver['urls'].items() if value[3] == url] if same_urls: continue number = urlserver['number'] # don't save urls already shortened if not url.startswith(urlserver_get_base_url()): urlserver['urls'][number] = ( datetime.datetime.now().strftime( urlserver_settings['http_time_format']), nick, buffer_short_name, url, '%s\t%s' % (prefix, message)) urls_short.append(urlserver_short_url(number)) if urlserver['buffer']: urlserver_display_url_detail(number) urlserver['number'] += 1 # remove old URLs if we have reach max list size urls_amount = 50 try: urls_amount = int(urlserver_settings['urls_amount']) if urls_amount <= 0: urls_amount = 50 except: urls_amount = 50 while len(urlserver['urls']) > urls_amount: keys = sorted(urlserver['urls']) del urlserver['urls'][keys[0]] return urls_short def urlserver_print_cb(data, buffer, time, tags, displayed, highlight, prefix, message): """ Callback for message printed in buffer: display short URLs after message. """ global urlserver, urlserver_settings if not displayed and urlserver_settings['msg_filtered'] != 'on': return weechat.WEECHAT_RC_OK if urlserver_settings['display_urls'] == 'on': buffer_full_name = '%s.%s' % ( weechat.buffer_get_string(buffer, 'plugin'), weechat.buffer_get_string(buffer, 'name')) if urlserver_settings['buffer_short_name'] == 'on': buffer_short_name = weechat.buffer_get_string(buffer, 'short_name') else: buffer_short_name = buffer_full_name urls_short = urlserver_update_urllist(buffer_full_name, buffer_short_name, tags, prefix, message) if urls_short: if urlserver_settings['separators'] and \ len(urlserver_settings['separators']) == 3: separator = ' %s ' % (urlserver_settings['separators'][1]) urls_string = separator.join(urls_short) urls_string = '%s %s %s' % ( urlserver_settings['separators'][0], urls_string, urlserver_settings['separators'][2]) else: urls_string = ' | '.join(urls_short) urls_string = '[ ' + urls_string + ' ]' weechat.prnt_date_tags( buffer, 0, 'no_log,notify_none', '%s%s' % (weechat.color(urlserver_settings['color']), urls_string)) return weechat.WEECHAT_RC_OK def urlserver_modifier_irc_cb(data, modifier, modifier_data, string): """Modifier for IRC message: add short URLs at the end of IRC message.""" global urlserver, urlserver_settings if urlserver_settings['display_urls_in_msg'] != 'on': return string msg = weechat.info_get_hashtable( 'irc_message_parse', {'message': string, 'server': modifier_data}) if 'nick' not in msg or 'channel' not in msg or 'arguments' not in msg: return string try: message = msg['arguments'].split(' ', 1)[1] if message.startswith(':'): message = message[1:] except: return string info_name = '%s,%s' % (modifier_data, msg['channel']) if weechat.info_get('irc_is_channel', info_name) == '1': name = msg['channel'] else: name = msg['nick'] buffer_full_name = 'irc.%s.%s' % (modifier_data, name) if urlserver_settings['buffer_short_name'] == 'on': buffer_short_name = name else: buffer_short_name = buffer_full_name urls_short = urlserver_update_urllist(buffer_full_name, buffer_short_name, None, msg['nick'], message, msg['nick']) if urls_short: if urlserver_settings['separators'] and \ len(urlserver_settings['separators']) == 3: separator = ' %s ' % (urlserver_settings['separators'][1]) urls_string = separator.join(urls_short) urls_string = '%s %s %s' % ( urlserver_settings['separators'][0], urls_string, urlserver_settings['separators'][2]) else: urls_string = ' | '.join(urls_short) urls_string = '[ ' + urls_string + ' ]' if urlserver_settings['color_in_msg']: urls_string = '\x03%s%s' % (urlserver_settings['color_in_msg'], urls_string) string = "%s %s" % (string, urls_string) return string def urlserver_config_cb(data, option, value): """Called when a script option is changed.""" global urlserver_settings pos = option.rfind('.') if pos > 0: name = option[pos+1:] if name in urlserver_settings: if name == 'http_allowed_ips': urlserver_settings[name] = re.compile(value) else: urlserver_settings[name] = value if name in ('http_hostname', 'http_port'): # don't restart if autostart is disabled and server isn't # already running if urlserver_settings['http_autostart'] == 'on' or \ urlserver['socket']: urlserver_server_restart() return weechat.WEECHAT_RC_OK def urlserver_filename(): """Return name of file used to store list of urls.""" options = { 'directory': 'data', } return weechat.string_eval_path_home('%h/urlserver_list.txt', {}, {}, options) def urlserver_read_urls(): """Read file with URLs.""" global urlserver filename = urlserver_filename() if os.path.isfile(filename): urlserver['number'] = 0 try: urlserver['urls'] = ast.literal_eval(open(filename, 'r').read()) keys = sorted(urlserver['urls']) if keys: urlserver['number'] = keys[-1] + 1 else: urlserver['number'] = 0 except: weechat.prnt('', '%surlserver: error reading file "%s"' % ( weechat.prefix('error'), filename)) def urlserver_write_urls(): """Write file with URLs.""" global urlserver keys = sorted(urlserver['urls']) content = '{\n%s\n}\n' % '\n'.join([ ' %d: %s,' % (key, str(urlserver['urls'][key])) for key in keys ]) open(urlserver_filename(), 'w').write(content) def urlserver_end(): """Script unloaded (oh no, why?)""" urlserver_server_stop() urlserver_write_urls() return weechat.WEECHAT_RC_OK if __name__ == '__main__' and import_ok: if weechat.register(SCRIPT_NAME, SCRIPT_AUTHOR, SCRIPT_VERSION, SCRIPT_LICENSE, SCRIPT_DESC, 'urlserver_end', ''): # set default settings version = weechat.info_get('version_number', '') or 0 for option, value in urlserver_settings_default.items(): if weechat.config_is_set_plugin(option): urlserver_settings[option] = weechat.config_get_plugin(option) else: weechat.config_set_plugin(option, value[0]) urlserver_settings[option] = value[0] if int(version) >= 0x00030500: weechat.config_set_desc_plugin( option, '%s (default: "%s")' % (value[1], value[0])) # detect config changes weechat.hook_config('plugins.var.python.%s.*' % SCRIPT_NAME, 'urlserver_config_cb', '') # add command weechat.hook_command( SCRIPT_COMMAND, SCRIPT_DESC, 'start|restart|stop|status || clear', ' start: start server\n' 'restart: restart server\n' ' stop: stop server\n' ' status: display status of server\n' ' clear: remove all URLs from list\n\n' 'Without argument, this command opens new buffer with list of ' 'URLs.\n\n' 'Initial setup:\n' ' - by default, script will listen on a random free port, ' 'you can force a port with:\n' ' /set plugins.var.python.urlserver.http_port "1234"\n' ' - you can force an IP or custom hostname with:\n' ' /set plugins.var.python.urlserver.http_hostname ' '"111.22.33.44"\n' ' - it is strongly recommended to restrict IPs allowed and/or ' 'use auth, for example:\n' ' /set plugins.var.python.urlserver.http_allowed_ips ' '"^(123.45.67.89|192.160.*)$"\n' ' /set plugins.var.python.urlserver.http_auth ' '"user:password"\n' ' - if you do not like the default HTML formatting, you can ' 'override the CSS:\n' ' /set plugins.var.python.urlserver.http_css_url ' '"http://example.com/sample.css"\n' ' See https://raw.github.com/FiXato/weechat_scripts/master/' 'urlserver/sample.css\n' ' - don\'t like the built-in HTTP server to start automatically? ' 'Disable it:\n' ' /set plugins.var.python.urlserver.http_autostart "off"\n' ' - have external port 80 or 443 (https) forwarded to your ' 'internal server port? Remove :port with:\n' ' /set plugins.var.python.urlserver.http_port_display "80" ' 'or "443" respectively\n' '\n' 'Tip: use URL without key at the end to display list of all URLs ' 'in your browser.', 'start|restart|stop|status|clear', 'urlserver_cmd_cb', '') if urlserver_settings['http_autostart'] == 'on': # start mini HTTP server urlserver_server_start() # load urls from file urlserver_read_urls() # catch URLs in buffers weechat.hook_print('', '', '://', 1, 'urlserver_print_cb', '') # modify URLS in irc messages (for relay) weechat.hook_modifier('irc_in2_privmsg', 'urlserver_modifier_irc_cb', '') weechat.hook_modifier('irc_in2_notice', 'urlserver_modifier_irc_cb', '') # search buffer urlserver['buffer'] = weechat.buffer_search('python', SCRIPT_BUFFER)