逆向解析 ColorOS 离线号码库 (二):多级指针寻址的运营商识别与数据自愈
0. 前言
在上一节中,我们成功逆向了号码归属地的 12-Bit 极限压缩算法。本节我们将继续深入,解析系统的运营商查询逻辑,并修复归属地查询中遗留的“异常区号”问题,最终实现一个归属地与运营商全息查询的完整脚本。
1. 架构分析:归属地与运营商的解耦
如果在归属地数据库(PhoneNumberData_3_1_0.dat)中寻找运营商信息,结果会一无所获。在底层的架构设计中,归属地与运营商采用了两套完全不同的存储与匹配逻辑:
- 归属地映射的离散性:一个
138开头的号码,其归属地可能遍布全国各地(取决于号码中间 4 位)。这种离散性无法通过简单的公式计算,因此必须依赖体积达数 MB 的二进制字典树(Trie)进行硬映射。 - 运营商分配的规律性:在“携号转网”政策全面实施前,中国大陆的手机运营商是严格按前 3 位(或前 4 位)号段分配的(例如
138归属移动,186归属联通)。因此,将规律性极强的数据与庞大的归属地数据库分离,可以有效节省存储空间并提升查询效率。
通过对代码的进一步反编译,我们定位到了两个专门处理运营商及特殊情况的文件:
CarrierData_1_0.dat:体积较小,存储了基础的号段与运营商对应关系。PortabilityNumberData.dat:携号转网(MNP)数据库,用于修正发生转网的特殊号码。
2. 解析 CarrierData 数据结构
public final void q() throws IOException { File file = new File(f920p.getFilesDir(), "CarrierData_1_0.dat"); if (!file.exists()) { e.b("OplusCarrierManager", "loadCarrierNameMap dataFile not exist"); return;59 collapsed lines
} DataInputStream dataInputStream = new DataInputStream(new FileInputStream(file)); try { dataInputStream.skipBytes(16); int unsignedShort = dataInputStream.readUnsignedShort(); for (int i3 = 0; i3 < unsignedShort; i3++) { int unsignedShort2 = dataInputStream.readUnsignedShort(); String[] strArr = new String[4]; for (int i4 = 0; i4 < 4; i4++) { strArr[i4] = BuildConfig.FLAVOR; } byte[] bArr = new byte[16]; for (int i5 = 0; i5 < 4; i5++) { int unsignedShort3 = dataInputStream.readUnsignedShort(); if (unsignedShort3 > 0) { if (unsignedShort3 > bArr.length) { bArr = new byte[unsignedShort3]; } dataInputStream.readFully(bArr, 0, unsignedShort3); strArr[i5] = new String(bArr, 0, unsignedShort3, l2.c.f761b); } } f917m.put(Integer.valueOf(unsignedShort2), strArr); } int unsignedShort4 = dataInputStream.readUnsignedShort(); for (int i6 = 0; i6 < unsignedShort4; i6++) { int i7 = dataInputStream.readInt(); int length = String.valueOf(i7).length(); f914j = Math.max(length, f914j); f915k = Math.min(length, f915k); f913i.put(Integer.valueOf(i7), Integer.valueOf(dataInputStream.readUnsignedShort())); } int unsignedShort5 = dataInputStream.readUnsignedShort(); for (int i8 = 0; i8 < unsignedShort5; i8++) { int i9 = dataInputStream.readInt(); int length2 = String.valueOf(i9).length(); f909e = Math.max(length2, f909e); f910f = Math.min(length2, f910f); f918n.put(Integer.valueOf(i9), Long.valueOf(dataInputStream.readInt())); } t1.k kVar = t1.k.f1147a; c2.a.a(dataInputStream, null); if (f906b) { e.a("OplusCarrierManager", "carrierNumberNameMap:"); for (Map.Entry<Integer, String[]> entry : f917m.entrySet()) { e.a("OplusCarrierManager", entry.getKey().intValue() + " : " + u1.j.l(entry.getValue())); } e.a("OplusCarrierManager", "networkNumberFileOffsetMap:" + f918n); } f911g = new ThreadLocal<>(); } catch (Throwable th) { try { throw th; } catch (Throwable th2) { c2.a.a(dataInputStream, th); throw th2; } }}通过反编译应用中的 loadCarrierNameMap(即代码中的 q() 方法),我们发现系统对运营商数据的存储采用了一套 “多级索引+内存偏移寻址” 的二进制文件结构。
根据 DataInputStream 的顺序读取逻辑,该 .dat 文件采用 大端序 (Big-Endian) 存储,主要分为四大区域:
- 文件头 (Header) - 16 字节:包含版本号和 CRC32 校验码,代码中通过
skipBytes(16)直接跳过。 - 运营商名称字典 (Carrier Names):首先读取运营商总数,随后依次读取对应的 ID 以及中文、英文、繁体台湾、繁体香港 4 个语言版本的名称(通过
readUnsignedShort获取字符串长度,再读取 UTF-8 字符串)。 - 全局通用号段表 (General Prefixes):对应代码中的
f913i变量。存放前 3 位或前 4 位即可直接决定运营商的基础号段。数据结构为一个 32 位 Int(前缀)对应一个 16 位 Short(运营商 ID)。 - 精细偏移映射表 (Detailed Offsets):对应代码中的
f918n变量。针对按万号段细分的复杂号码(如虚拟运营商),该表记录了号段前缀在文件底部的绝对字节偏移量 (Offset)。
2.1 双轨查询逻辑
在实际查询时,系统设计了两条路径:
private final String t(String str, int i3) { int i4 = f909e; int i5 = f910f; if (i5 <= i4) { while (true) {32 collapsed lines
String strSubstring = str.substring(0, i4); i.d(strSubstring, "substring(...)"); Long l3 = f918n.get(Integer.valueOf(Integer.parseInt(strSubstring))); if (l3 == null) { if (i4 == i5) { break; } i4--; } else { String strSubstring2 = str.substring(i4, i4 + 4); i.d(strSubstring2, "substring(...)"); String strK = k(i3, strSubstring, strSubstring2); c<String, String> cVar = f916l; String strD = cVar.d(strK); if (strD != null) { if (f906b) { e.c("OplusCarrierManager", str + " cache hit "); } return strD; } String[] strArr = f917m.get(Integer.valueOf(r(Long.valueOf(l3.longValue() + ((long) (Integer.parseInt(strSubstring2) * 2))).longValue()))); if (strArr == null) { cVar.e(strK, BuildConfig.FLAVOR); return null; } String str2 = (String) u1.j.h(strArr, i3); return TextUtils.isEmpty(str2) ? (String) u1.j.h(strArr, 0) : str2; } } } return null;}- 精确路径 (
t()方法):处理新号段或混合号段(如1705虚拟运营商)。光匹配前几位并不足够,系统会查出前缀对应的底层偏移量,再加上号码中段数字计算出内存指针,直接跳转到文件尾部读取真实的运营商 ID。
private final String u(String str, int i3) { int i4 = f914j; int i5 = f915k; if (i5 <= i4) { while (true) { String strSubstring = str.substring(0, i4);18 collapsed lines
i.d(strSubstring, "substring(...)"); Integer num = f913i.get(Integer.valueOf(Integer.parseInt(strSubstring))); if (num == null) { if (i4 == i5) { break; } i4--; } else { String[] strArr = f917m.get(num); if (strArr != null) { return (String) u1.j.h(strArr, i3); } return null; } } } return null;}- 快速路径 (
u()方法):处理传统号段(如138)。前三位即可 100% 确定运营商,系统只需在general_prefixes哈希表中查询一次即可迅速返回结果。
2.3 Python 算法重构
基于以上的逆向分析,我们完全脱离了 Android 环境,使用纯 Python 实现了对该二进制数据文件的脱机解析与查询。完整代码如下:
import osimport struct
def load_carrier_database(file_path): print("⏳ 正在加载并解析运营商底层数据库...")113 collapsed lines
with open(file_path, 'rb') as f: data = f.read()
offset = 16 # dataInputStream.skipBytes(16)
# 1. 读取运营商名称字典 carrier_names = {} num_carriers = struct.unpack_from(">H", data, offset)[0] offset += 2
for _ in range(num_carriers): carrier_id = struct.unpack_from(">H", data, offset)[0] offset += 2
names = [] # 依次读取 中文、英文、繁体台湾、繁体香港 4 个语言版本 for _ in range(4): str_len = struct.unpack_from(">H", data, offset)[0] offset += 2 if str_len > 0: # UTF-8 解码字符串 name = data[offset : offset + str_len].decode('utf-8', errors='ignore') offset += str_len else: name = "" names.append(name)
carrier_names[carrier_id] = names
# 2. 读取全局通用号段表 (f913i) general_prefixes = {} num_gen_prefixes = struct.unpack_from(">H", data, offset)[0] offset += 2
for _ in range(num_gen_prefixes): prefix_val = struct.unpack_from(">i", data, offset)[0] offset += 4 cid = struct.unpack_from(">H", data, offset)[0] offset += 2 general_prefixes[str(prefix_val)] = cid
# 3. 读取精细偏移映射表 (f918n) detailed_prefixes = {} num_det_prefixes = struct.unpack_from(">H", data, offset)[0] offset += 2
for _ in range(num_det_prefixes): prefix_val = struct.unpack_from(">i", data, offset)[0] offset += 4 file_offset = struct.unpack_from(">i", data, offset)[0] offset += 4 detailed_prefixes[str(prefix_val)] = file_offset
print(f"✅ 成功解析: {len(carrier_names)} 个运营商, {len(general_prefixes)} 个全局号段, {len(detailed_prefixes)} 个精细号段偏移")
return data, carrier_names, general_prefixes, detailed_prefixes
def query_carrier(phone, data, carrier_names, general_prefixes, detailed_prefixes): if len(phone) < 7 or not phone.isdigit(): return "❌ 请输入至少 7 位数字号码"
# 还原代码中的 t() 方法:优先匹配精细偏移映射 # 动态尝试长度 4 和 3 (反编译代码里的 f909e 到 f910f) for length in range(4, 2, -1): prefix = phone[:length] if prefix in detailed_prefixes: suffix = phone[length : length+4] if len(suffix) == 4: # 计算文件底部的绝对寻址: Offset + int(suffix) * 2 bytes target_offset = detailed_prefixes[prefix] + (int(suffix) * 2) if target_offset + 2 <= len(data): cid = struct.unpack_from(">H", data, target_offset)[0] if cid in carrier_names: # 返回索引 0 的中文名称 name = carrier_names[cid][0] return name if name else carrier_names[cid][1]
# 还原代码中的 u() 方法:匹配全局通用号段 for length in range(4, 2, -1): prefix = phone[:length] if prefix in general_prefixes: cid = general_prefixes[prefix] if cid in carrier_names: name = carrier_names[cid][0] return name if name else carrier_names[cid][1]
return "未知运营商"
def main(): file_name = "CarrierData_1_0.dat" if not os.path.exists(file_name): print(f"❌ 找不到文件:{file_name},请确保与脚本同目录。") return
data, carrier_names, general_prefixes, detailed_prefixes = load_carrier_database(file_name)
print("\n" + "="*40) print("📡 OPPO 底层运营商查询系统") print("="*40)
while True: try: user_input = input("\n👉 请输入 7位及以上手机号 (输入 q 退出): ").strip() if user_input.lower() == 'q': break
carrier = query_carrier(user_input, data, carrier_names, general_prefixes, detailed_prefixes) print(f"🏢 运营商: 【{carrier}】")
except KeyboardInterrupt: break except Exception as e: print(f"发生错误: {e}")3. 修复长途区号 “9999” 的墓碑机制
在初步运行上一节的归属地查询脚本时,部分地区的归属地虽然正确,但长途区号会异常显示为 9999。
这一现象源于近年来中国部分省份的行政区划调整和长途区号合并。例如:
- 广西南宁与崇左合并了区号
0771。 - 四川成都、眉山、资阳合并了区号
028。 - 辽宁沈阳、抚顺、铁岭合并了区号
024。
为了避免重新生成并下发数 MB 的底层二进制结构,工程师采用了一种巧妙的 “墓碑机制 (Tombstone)” :
- 将原有的老条目(如独立的“广西南宁”)区号修改为
9999,作为废弃标记。 - 在城市字典列表的末尾,追加新的合并城市名(如
0771 广西南宁/崇左)。
同时,系统引入了一个外挂补丁文件 Multi_Areano_Table.txt。该文件记录了新旧数据的映射关系:
origin_id cityname equal_id equal_cityname10 辽宁沈阳 379 辽宁沈阳/抚顺/铁岭13 四川成都 378 四川成都/眉山/资阳遗憾的是,当试图查看最核心的重定向方法 h() 时,Jadx 再次放弃了转换。无奈之下,只能继续硬着头皮去阅读底层的 Smali 汇编指令。
.method public h(Landroid/database/sqlite/SQLiteDatabase;)V .registers 11
.line 1 const-string v0, "PhoneNoDbHelper"438 collapsed lines
.line 2 .line 3 const/4 v1, 0x0
.line 4 :try_start_3 iget-object p0, p0, Ll0/d;->d:Landroid/content/Context;
.line 5 .line 6 invoke-virtual {p0}, Landroid/content/Context;->getAssets()Landroid/content/res/AssetManager;
.line 7 .line 8 .line 9 move-result-object p0
.line 10 const-string v2, "Multi_Areano_Table.txt"
.line 11 .line 12 invoke-virtual {p0, v2}, Landroid/content/res/AssetManager;->open(Ljava/lang/String;)Ljava/io/InputStream;
.line 13 .line 14 .line 15 move-result-object p0 :try_end_f .catch Ljava/lang/Exception; {:try_start_3 .. :try_end_f} :catch_a3 .catchall {:try_start_3 .. :try_end_f} :catchall_a0
.line 16 if-eqz p0, :cond_8e
.line 17 .line 18 if-nez p1, :cond_15
.line 19 .line 20 goto/16 :goto_8e
.line 21 .line 22 :cond_15 :try_start_15 invoke-static {}, Landroid/os/SystemClock;->elapsedRealtime()J
.line 23 .line 24 .line 25 move-result-wide v2
.line 26 new-instance v4, Ljava/io/BufferedReader;
.line 27 .line 28 new-instance v5, Ljava/io/InputStreamReader;
.line 29 .line 30 invoke-direct {v5, p0}, Ljava/io/InputStreamReader;-><init>(Ljava/io/InputStream;)V
.line 31 .line 32 .line 33 invoke-direct {v4, v5}, Ljava/io/BufferedReader;-><init>(Ljava/io/Reader;)V :try_end_23 .catch Ljava/lang/Exception; {:try_start_15 .. :try_end_23} :catch_9d .catchall {:try_start_15 .. :try_end_23} :catchall_99
.line 34 .line 35 .line 36 const/4 v1, 0x1
.line 37 :cond_24 :goto_24 :try_start_24 invoke-virtual {v4}, Ljava/io/BufferedReader;->readLine()Ljava/lang/String;
.line 38 .line 39 .line 40 move-result-object v5
.line 41 invoke-static {v5}, Landroid/text/TextUtils;->isEmpty(Ljava/lang/CharSequence;)Z
.line 42 .line 43 .line 44 move-result v6
.line 45 if-eqz v6, :cond_2f
.line 46 .line 47 goto :goto_39
.line 48 :cond_2f invoke-virtual {v5}, Ljava/lang/String;->trim()Ljava/lang/String;
.line 49 .line 50 .line 51 move-result-object v5
.line 52 invoke-static {v5}, Landroid/text/TextUtils;->isEmpty(Ljava/lang/CharSequence;)Z
.line 53 .line 54 .line 55 move-result v6
.line 56 if-eqz v6, :cond_56
.line 57 .line 58 :goto_39 new-instance p1, Ljava/lang/StringBuilder;
.line 59 .line 60 invoke-direct {p1}, Ljava/lang/StringBuilder;-><init>()V
.line 61 .line 62 .line 63 const-string v1, "updateCityCode use "
.line 64 .line 65 invoke-virtual {p1, v1}, Ljava/lang/StringBuilder;->append(Ljava/lang/String;)Ljava/lang/StringBuilder;
.line 66 .line 67 .line 68 invoke-static {}, Landroid/os/SystemClock;->elapsedRealtime()J
.line 69 .line 70 .line 71 move-result-wide v5
.line 72 sub-long/2addr v5, v2
.line 73 invoke-virtual {p1, v5, v6}, Ljava/lang/StringBuilder;->append(J)Ljava/lang/StringBuilder;
.line 74 .line 75 .line 76 invoke-virtual {p1}, Ljava/lang/StringBuilder;->toString()Ljava/lang/String;
.line 77 .line 78 .line 79 move-result-object p1
.line 80 invoke-static {v0, p1}, Landroid/util/Log;->d(Ljava/lang/String;Ljava/lang/String;)I :try_end_52 .catch Ljava/lang/Exception; {:try_start_24 .. :try_end_52} :catch_9e .catchall {:try_start_24 .. :try_end_52} :catchall_8c
.line 81 .line 82 .line 83 invoke-virtual {p0}, Ljava/io/InputStream;->close()V
.line 84 .line 85 .line 86 goto :goto_b0
.line 87 :cond_56 const/4 v6, 0x0
.line 88 if-eqz v1, :cond_5b
.line 89 .line 90 move v1, v6
.line 91 goto :goto_24
.line 92 :cond_5b :try_start_5b const-string v7, "\t"
.line 93 .line 94 invoke-virtual {v5, v7}, Ljava/lang/String;->split(Ljava/lang/String;)[Ljava/lang/String;
.line 95 .line 96 .line 97 move-result-object v5
.line 98 array-length v7, v5
.line 99 const/4 v8, 0x4
.line 100 if-ne v7, v8, :cond_24
.line 101 .line 102 new-instance v7, Ljava/lang/StringBuilder;
.line 103 .line 104 invoke-direct {v7}, Ljava/lang/StringBuilder;-><init>()V
.line 105 .line 106 .line 107 const-string v8, "UPDATE areano_and_citynames SET equal_id = \'"
.line 108 .line 109 invoke-virtual {v7, v8}, Ljava/lang/StringBuilder;->append(Ljava/lang/String;)Ljava/lang/StringBuilder;
.line 110 .line 111 .line 112 const/4 v8, 0x2
.line 113 aget-object v8, v5, v8
.line 114 .line 115 invoke-virtual {v7, v8}, Ljava/lang/StringBuilder;->append(Ljava/lang/String;)Ljava/lang/StringBuilder;
.line 116 .line 117 .line 118 const-string v8, "\' WHERE _id = \'"
.line 119 .line 120 invoke-virtual {v7, v8}, Ljava/lang/StringBuilder;->append(Ljava/lang/String;)Ljava/lang/StringBuilder;
.line 121 .line 122 .line 123 aget-object v5, v5, v6
.line 124 .line 125 invoke-virtual {v7, v5}, Ljava/lang/StringBuilder;->append(Ljava/lang/String;)Ljava/lang/StringBuilder;
.line 126 .line 127 .line 128 const-string v5, "\';"
.line 129 .line 130 invoke-virtual {v7, v5}, Ljava/lang/StringBuilder;->append(Ljava/lang/String;)Ljava/lang/StringBuilder;
.line 131 .line 132 .line 133 invoke-virtual {v7}, Ljava/lang/StringBuilder;->toString()Ljava/lang/String;
.line 134 .line 135 .line 136 move-result-object v5
.line 137 invoke-virtual {p1, v5}, Landroid/database/sqlite/SQLiteDatabase;->execSQL(Ljava/lang/String;)V :try_end_8b .catch Ljava/lang/Exception; {:try_start_5b .. :try_end_8b} :catch_9e .catchall {:try_start_5b .. :try_end_8b} :catchall_8c
.line 138 .line 139 .line 140 goto :goto_24
.line 141 :catchall_8c move-exception p1
.line 142 goto :goto_9b
.line 143 :cond_8e :goto_8e :try_start_8e const-string p1, "updateCityCode inputStream or db is null"
.line 144 .line 145 invoke-static {v0, p1}, Landroid/util/Log;->d(Ljava/lang/String;Ljava/lang/String;)I :try_end_93 .catch Ljava/lang/Exception; {:try_start_8e .. :try_end_93} :catch_9d .catchall {:try_start_8e .. :try_end_93} :catchall_99
.line 146 .line 147 .line 148 if-eqz p0, :cond_98
.line 149 .line 150 invoke-virtual {p0}, Ljava/io/InputStream;->close()V
.line 151 .line 152 .line 153 :cond_98 return-void
.line 154 :catchall_99 move-exception p1
.line 155 move-object v4, v1
.line 156 :goto_9b move-object v1, p0
.line 157 goto :goto_b5
.line 158 :catch_9d move-object v4, v1
.line 159 :catch_9e move-object v1, p0
.line 160 goto :goto_a4
.line 161 :catchall_a0 move-exception p1
.line 162 move-object v4, v1
.line 163 goto :goto_b5
.line 164 :catch_a3 move-object v4, v1
.line 165 :goto_a4 :try_start_a4 const-string p0, "update city code error"
.line 166 .line 167 invoke-static {v0, p0}, Landroid/util/Log;->e(Ljava/lang/String;Ljava/lang/String;)I :try_end_a9 .catchall {:try_start_a4 .. :try_end_a9} :catchall_b4
.line 168 .line 169 .line 170 if-eqz v1, :cond_ae
.line 171 .line 172 invoke-virtual {v1}, Ljava/io/InputStream;->close()V
.line 173 .line 174 .line 175 :cond_ae if-eqz v4, :cond_b3
.line 176 .line 177 :goto_b0 invoke-virtual {v4}, Ljava/io/BufferedReader;->close()V
.line 178 .line 179 .line 180 :cond_b3 return-void
.line 181 :catchall_b4 move-exception p1
.line 182 :goto_b5 if-eqz v1, :cond_ba
.line 183 .line 184 invoke-virtual {v1}, Ljava/io/InputStream;->close()V
.line 185 .line 186 .line 187 :cond_ba if-eqz v4, :cond_bf
.line 188 .line 189 invoke-virtual {v4}, Ljava/io/BufferedReader;->close()V
.line 190 .line 191 .line 192 :cond_bf throw p1.end method在对应的 Smali 代码中,系统在初始化 SQLite 数据库时执行了类似如下的重定向操作:
-- 省略部分逻辑UPDATE areano_and_citynames SET equal_id = '379' WHERE _id = '10';这意味着,当底层解密算出号码属于旧的 id = 10 时,系统会通过重定向直接获取 id = 379 的合并后数据。
4. 全息解析 Python 脚本整合
结合以上两节的分析,我们将官方的 Multi_Areano_Table.txt 补丁动态加载至内存,并融合了多级指针寻址的运营商识别机制,实现了归属地与运营商的“二合一”查询。
以下是完整的 Python 实现代码:
import osimport struct
# ==========================================# 模块 1:归属地数据库解析 (PhoneNumberData)217 collapsed lines
# ==========================================def load_location_db(file_path): print("⏳ [1/2] 正在加载 归属地 数据库...") with open(file_path, 'rb') as f: data = f.read()
tail_data = data[-12002:] area_codes_raw = tail_data[2:2002] city_names_raw = tail_data[2002:10002] extend_prefix_raw = tail_data[10002:12002]
cities = [] for i in range(400): area = area_codes_raw[i*5 : i*5+5].decode('ascii', errors='ignore').strip() name_bytes = city_names_raw[i*20 : i*20+20].split(b'\x00')[0] name = name_bytes.decode('gbk', errors='ignore').strip() cities.append([area, name])
# 🌟 官方补丁外挂:动态加载 Multi_Areano_Table.txt 处理重定向 patch_file = "Multi_Areano_Table.txt" if os.path.exists(patch_file): print(" -> 发现官方区号修正补丁,正在进行数据合并重定向...") with open(patch_file, "r", encoding="utf-8") as f: lines = f.readlines() for line in lines[1:]: # 跳过表头 parts = line.strip().split('\t') if len(parts) == 4: origin_id = int(parts[0]) equal_id = int(parts[2])
# 数据库 _id 从 1 开始,数组索引需减 1 target_area, target_name = cities[equal_id - 1] # 覆盖旧的 9999 废弃数据 cities[origin_id - 1][0] = target_area else: print(f" -> ⚠️ 未找到 {patch_file},部分合并城市的区号可能显示为 9999。")
# 构建号段偏移映射表 prefix_map = {} base_idx = 0 for p in range(130, 140): prefix_map[str(p)] = base_idx; base_idx += 10000 for p in range(150, 160): prefix_map[str(p)] = base_idx; base_idx += 10000 prefix_map["188"] = base_idx; base_idx += 10000 prefix_map["189"] = base_idx; base_idx += 10000
for i in range(1000): b1 = extend_prefix_raw[i*2] b2 = extend_prefix_raw[i*2 + 1] prefix_val = (b1 << 8) | b2 if prefix_val == 0: break prefix_map[str(prefix_val)] = base_idx base_idx += 10000
binary_array = data[16 : -12002] print(f"✅ [1/2] 归属地加载成功!(解析号段数: {len(prefix_map)})")
return prefix_map, cities, binary_array
# ==========================================# 模块 2:运营商数据库解析 (CarrierData)# ==========================================def load_carrier_db(file_path): print("⏳ [2/2] 正在加载 运营商 数据库...") with open(file_path, 'rb') as f: data = f.read()
offset = 16
# 1. 解析运营商名称字典 carrier_names = {} num_carriers = struct.unpack_from(">H", data, offset)[0] offset += 2 for _ in range(num_carriers): carrier_id = struct.unpack_from(">H", data, offset)[0] offset += 2 names = [] for _ in range(4): str_len = struct.unpack_from(">H", data, offset)[0] offset += 2 if str_len > 0: name = data[offset : offset + str_len].decode('utf-8', errors='ignore') offset += str_len else: name = "" names.append(name) carrier_names[carrier_id] = names
# 2. 解析全局通用号段表 general_prefixes = {} num_gen_prefixes = struct.unpack_from(">H", data, offset)[0] offset += 2 for _ in range(num_gen_prefixes): prefix_val = struct.unpack_from(">i", data, offset)[0] offset += 4 cid = struct.unpack_from(">H", data, offset)[0] offset += 2 general_prefixes[str(prefix_val)] = cid
# 3. 解析精细偏移映射表 detailed_prefixes = {} num_det_prefixes = struct.unpack_from(">H", data, offset)[0] offset += 2 for _ in range(num_det_prefixes): prefix_val = struct.unpack_from(">i", data, offset)[0] offset += 4 file_offset = struct.unpack_from(">i", data, offset)[0] offset += 4 detailed_prefixes[str(prefix_val)] = file_offset
print(f"✅ [2/2] 运营商加载成功!(精细号段规则: {len(detailed_prefixes)}条)") return data, carrier_names, general_prefixes, detailed_prefixes
# ==========================================# 模块 3:全息信息统一查询接口# ==========================================def query_phone_info(phone, loc_data, car_data): if len(phone) < 7 or not phone.isdigit(): return "❌ 格式错误:请输入至少 7 位纯数字号码!"
eval_phone = phone[:8] phone_7 = eval_phone[:7]
# --- A. 归属地查询逻辑 (12-Bit 压缩解密) --- prefix_map, cities, binary_array = loc_data prefix = phone_7[:3] suffix = phone_7[3:]
loc_name = "未知" area_code = "未知"
if prefix in prefix_map: absolute_index = prefix_map[prefix] + int(suffix) byte_offset = (absolute_index // 2) * 3
if byte_offset + 2 < len(binary_array): b1 = binary_array[byte_offset] b2 = binary_array[byte_offset + 1] b3 = binary_array[byte_offset + 2]
if absolute_index % 2 == 0: city_id = (b1 << 4) | (b2 >> 4) else: city_id = ((b2 & 0x0F) << 8) | b3
if city_id != 0 and city_id < len(cities): area_code, loc_name = cities[city_id] if not loc_name: loc_name = "未知"
# --- B. 运营商查询逻辑 (多级指针寻址) --- car_raw_data, carrier_names, general_prefixes, detailed_prefixes = car_data carrier_name = "未知运营商" found_carrier = False
# B.1 优先进行精细规则匹配 for length in range(4, 2, -1): p_prefix = eval_phone[:length] if p_prefix in detailed_prefixes: p_suffix = eval_phone[length : length+4] if len(p_suffix) == 4: target_offset = detailed_prefixes[p_prefix] + (int(p_suffix) * 2) if target_offset + 2 <= len(car_raw_data): cid = struct.unpack_from(">H", car_raw_data, target_offset)[0] if cid in carrier_names: carrier_name = carrier_names[cid][0] or carrier_names[cid][1] found_carrier = True break
# B.2 若未命中,则进行全局大号段匹配 if not found_carrier: for length in range(4, 2, -1): p_prefix = eval_phone[:length] if p_prefix in general_prefixes: cid = general_prefixes[p_prefix] if cid in carrier_names: carrier_name = carrier_names[cid][0] or carrier_names[cid][1] break
result = ( f"📱 查询号码: {phone}\n" f"📍 归属地点: 【{loc_name}】 (长途区号: {area_code})\n" f"🏢 运营网络: 【{carrier_name}】" ) return result
def main(): print("=" * 45) print("🚀 ColorOS 离线号码全息识别系统") print("=" * 45)
loc_file = "PhoneNumberData_3_1_0.dat" car_file = "CarrierData_1_0.dat"
if not os.path.exists(loc_file) or not os.path.exists(car_file): print(f"❌ 启动失败:缺少数据库文件。\n请确保 {loc_file} 和 {car_file} 都在同一目录!") return
loc_data = load_location_db(loc_file) car_data = load_carrier_db(car_file) print("-" * 45)
while True: try: user_input = input("\n👉 请输入完整的手机号 (输入 q 退出): ").strip() if user_input.lower() == 'q': print("👋 系统已退出。") break
result = query_phone_info(user_input, loc_data, car_data) print("-" * 35) print(result) print("-" * 35)
except KeyboardInterrupt: print("\n👋 系统已退出。") break except Exception as e: print(f"⚠️ 查询中发生异常: {e}")5. 小结
至此,我们已经成功提取并重构了基于号码前 7 位的归属地与运营商查询逻辑。但对于发生过“携号转网”的特例号码,传统的号段查询便会失效。在下一节中,我们将解析 PortabilityNumberData.dat 特权库,探究系统是如何利用二分查找算法解决携号转网难题的,并最终实现支持高并发的数据全量导出脚本。