1#!/usr/bin/python3 2 3import re 4import json 5import logging 6import sys 7from collections import defaultdict, namedtuple 8 9AnopeObject = namedtuple('AnopeObject', ('type', 'kv')) 10 11MASK_MAGIC_REGEX = re.compile(r'[*?!@]') 12 13def access_level_to_amode(level): 14 # https://wiki.anope.org/index.php/2.0/Modules/cs_xop 15 if level == 'QOP': 16 return 'q' 17 elif level == 'SOP': 18 return 'a' 19 elif level == 'AOP': 20 return 'o' 21 elif level == 'HOP': 22 return 'h' 23 elif level == 'VOP': 24 return 'v' 25 26 try: 27 level = int(level) 28 except: 29 return None 30 if level >= 10000: 31 return 'q' 32 elif level >= 9999: 33 return 'a' 34 elif level >= 5: 35 return 'o' 36 elif level >= 4: 37 return 'h' 38 elif level >= 3: 39 return 'v' 40 else: 41 return None 42 43def to_unixnano(timestamp): 44 return int(timestamp) * (10**9) 45 46def file_to_objects(infile): 47 result = [] 48 obj = None 49 for line in infile: 50 pieces = line.rstrip('\r\n').split(' ', maxsplit=2) 51 if len(pieces) == 0: 52 logging.warning("skipping blank line in db") 53 continue 54 if pieces[0] == 'END': 55 result.append(obj) 56 obj = None 57 elif pieces[0] == 'OBJECT': 58 obj = AnopeObject(pieces[1], {}) 59 elif pieces[0] == 'DATA': 60 obj.kv[pieces[1]] = pieces[2] 61 else: 62 raise ValueError("unknown command found in anope db", pieces[0]) 63 return result 64 65ANOPE_MODENAME_TO_MODE = { 66 'NOEXTERNAL': 'n', 67 'TOPIC': 't', 68 'INVITE': 'i', 69 'NOCTCP': 'C', 70 'AUDITORIUM': 'u', 71 'SECRET': 's', 72} 73 74def convert(infile): 75 out = { 76 'version': 1, 77 'source': 'anope', 78 'users': defaultdict(dict), 79 'channels': defaultdict(dict), 80 } 81 82 objects = file_to_objects(infile) 83 84 lastmode_channels = set() 85 86 for obj in objects: 87 if obj.type == 'NickCore': 88 username = obj.kv['display'] 89 userdata = {'name': username, 'hash': obj.kv['pass'], 'email': obj.kv['email']} 90 out['users'][username] = userdata 91 elif obj.type == 'NickAlias': 92 username = obj.kv['nc'] 93 nick = obj.kv['nick'] 94 userdata = out['users'][username] 95 if username.lower() == nick.lower(): 96 userdata['registeredAt'] = to_unixnano(obj.kv['time_registered']) 97 else: 98 if 'additionalNicks' not in userdata: 99 userdata['additionalNicks'] = [] 100 userdata['additionalNicks'].append(nick) 101 elif obj.type == 'ChannelInfo': 102 chname = obj.kv['name'] 103 founder = obj.kv['founder'] 104 chdata = { 105 'name': chname, 106 'founder': founder, 107 'registeredAt': to_unixnano(obj.kv['time_registered']), 108 'topic': obj.kv['last_topic'], 109 'topicSetBy': obj.kv['last_topic_setter'], 110 'topicSetAt': to_unixnano(obj.kv['last_topic_time']), 111 'amode': {founder: 'q',} 112 } 113 # DATA last_modes INVITE KEY,hunter2 NOEXTERNAL REGISTERED TOPIC 114 last_modes = obj.kv.get('last_modes') 115 if last_modes: 116 modes = [] 117 for mode_desc in last_modes.split(): 118 if ',' in mode_desc: 119 mode_name, mode_value = mode_desc.split(',', maxsplit=1) 120 else: 121 mode_name, mode_value = mode_desc, None 122 if mode_name == 'KEY': 123 chdata['key'] = mode_value 124 else: 125 modes.append(ANOPE_MODENAME_TO_MODE.get(mode_name, '')) 126 chdata['modes'] = ''.join(modes) 127 # prevent subsequent ModeLock objects from modifying the mode list further: 128 lastmode_channels.add(chname) 129 out['channels'][chname] = chdata 130 elif obj.type == 'ModeLock': 131 if obj.kv.get('set') != '1': 132 continue 133 chname = obj.kv['ci'] 134 if chname in lastmode_channels: 135 continue 136 chdata = out['channels'][chname] 137 modename = obj.kv['name'] 138 if modename == 'KEY': 139 chdata['key'] = obj.kv['param'] 140 else: 141 oragono_mode = ANOPE_MODENAME_TO_MODE.get(modename) 142 if oragono_mode is not None: 143 stored_modes = chdata.get('modes', '') 144 stored_modes += oragono_mode 145 chdata['modes'] = stored_modes 146 elif obj.type == 'ChanAccess': 147 chname = obj.kv['ci'] 148 target = obj.kv['mask'] 149 mode = access_level_to_amode(obj.kv['data']) 150 if mode is None: 151 continue 152 if MASK_MAGIC_REGEX.search(target): 153 continue 154 chdata = out['channels'][chname] 155 amode = chdata.setdefault('amode', {}) 156 amode[target] = mode 157 chdata['amode'] = amode 158 159 # do some basic integrity checks 160 for chname, chdata in out['channels'].items(): 161 founder = chdata.get('founder') 162 if founder not in out['users']: 163 raise ValueError("no user corresponding to channel founder", chname, chdata.get('founder')) 164 165 return out 166 167def main(): 168 if len(sys.argv) != 3: 169 raise Exception("Usage: anope2json.py anope.db output.json") 170 with open(sys.argv[1]) as infile: 171 output = convert(infile) 172 with open(sys.argv[2], 'w') as outfile: 173 json.dump(output, outfile) 174 175if __name__ == '__main__': 176 logging.basicConfig() 177 sys.exit(main()) 178