LINE macOS .edb → SQLite decryptor + LINE-style chat reader web UI (wxSQLite3 AES-128-CBC codec)

Starred Gists，GitHub Gist，即时分享代码、笔记与代码片段。关注

LINE macOS .edb → SQLite decryptor + chat reader

Decrypt LINE Desktop (macOS) encrypted chat databases (.edb) into plain SQLite, then browse them with a LINE-style web UI.

Works on LINE 26.5.0 (Qt build, Mac App Store). Educational/forensic use on your own data only.

How it works

LINE Mac encrypts its SQLite DBs with a wxSQLite3-derived codec:

  • passphrase — 32 hex chars, synced from the server, lives only in process memory (never on disk/keychain). Capture it once with an LLDB trap on a re-signed debuggable copy (see below), or reuse a known one.
  • base key — PDF-style derivation: pad → MD5×50 → RC4×20 → MD5×50
  • per page (4096 B, AES-128-CBC):
    • pagekey(p) = MD5(base ‖ LE32(p) ‖ "sAlT")
    • iv(p) = MD5(LCG(p+1) mod 0x7FFFFFC7) — older app versions wrote pages with modulus 0x7FFFFF07; the decryptor auto-detects per page via the SQLite b-tree type byte.
  • page 1 stores a split header: decrypt file[8:16] ‖ file[24:4096] and prepend "SQLite format 3\0". The file magic FC AA 0B C6 CC 69 FB C7 is literally AES(pk₁,iv₁,"SQLite format 3\0")[0:8].

Usage

# 1) decrypt (needs the passphrase for your account)
python3 -m venv .venv && .venv/bin/pip install pycryptodome
.venv/bin/python decrypt_line.py ~/Library/Containers/jp.naver.line.mac/Data/Library/Containers/jp.naver.line/Data/db/q.edb main.sqlite 

# 2) browse
.venv/bin/python server.py main.sqlite 8819
open http://127.0.0.1:8819

Capturing the passphrase (one-time, per account)

  1. Copy the app, delete _MASReceipt, strip team entitlements + build rpaths, re-sign adhoc with only get-task-allow + com.apple.security.cs.disable-library-validation.
  2. Attach LLDB with a breakpoint at the codec's pad_password/derive_key (LINE 26.5.0 arm64: 0x101b39f28 / 0x101b3a04c) logging register x1.
  3. Log in once (QR) in that copy; the 32-hex string in the trap log is the passphrase — it also decrypts the original sandboxed account DBs.

Reader

server.py (stdlib-only, read-only SQLite) + index.html: chat list with last message previews, LINE-style bubbles (green outgoing / white incoming), group sender names, date separators, message search, infinite scroll. Tables used: _message, _chat, _contact, _groupChat.

#!/usr/bin/env python3
"""Decrypt a LINE macOS .edb chat database to plain SQLite.
Usage: python decrypt_line.py <32-hex-passphrase>
Codec (LINE 26.5.0, wxSQLite3-derived):
base_key = MD5^50(RC4^20(pad(passphrase))) chain starting from pad("")
pagekey = MD5(base_key || LE32(page) || b"sAlT")
iv = MD5(LCG(page+1) mod 0x7FFFFFC7) [older pages: mod 0x7FFFFF07]
cipher = AES-128-CBC, 4096-byte pages
page 1 = "SQLite format 3\0" || CBC-dec(file[8:16] || file[24:4096])
"""
import struct
import sys
from Crypto.Cipher import AES
from Crypto.Hash import MD5
PDF_PADDING = bytes.fromhex(
"28bf4e5e4e758a4164004e56fffa0108"
"2e2e00b6d0683e802f0ca9fe6453697a"
)
BTREE_TYPES = (2, 5, 10, 13)
IV_MODULI = (0x7FFFFFC7, 0x7FFFFF07) # new generation, old generation
def md5(data: bytes) -> bytes:
return MD5.new(data).digest()
def pad_password(pw: str) -> bytes:
pb = pw.encode()
if len(pb) >= 32:
return pb[:32]
return pb + PDF_PADDING[len(pb):]
def rc4(key: bytes, data: bytes) -> bytes:
s = list(range(256))
j = 0
for i in range(256):
j = (j + s[i] + key[i % len(key)]) & 0xFF
s[i], s[j] = s[j], s[i]
i = j = 0
out = bytearray()
for b in data:
i = (i + 1) & 0xFF
j = (j + s[i]) & 0xFF
s[i], s[j] = s[j], s[i]
out.append(b ^ s[(s[i] + s[j]) & 0xFF])
return bytes(out)
def derive_base_key(passphrase: str) -> bytes:
user_pad = pad_password(passphrase)
owner_pad = pad_password("")
digest = md5(owner_pad)
for _ in range(50):
digest = md5(digest)
owner_key = bytearray(user_pad)
for rnd in range(20):
digest_r = bytes(b ^ rnd for b in digest)
owner_key = bytearray(rc4(digest_r, bytes(owner_key)))
digest = md5(user_pad + bytes(owner_key))
for _ in range(50):
digest = md5(digest)
return digest[:16]
def _lcg(z: int, modulus: int) -> int:
q = z // 52774
z = 40692 * (z - 52774 * q) - 3791 * q
return z + modulus if z < 0 else z
def page_iv(page: int, modulus: int) -> bytes:
z = page + 1
buf = bytearray(16)
for j in range(4):
z = _lcg(z, modulus)
buf[4 * j:4 * j + 4] = struct.pack("<I", z)
return md5(bytes(buf))
def page_key(base: bytes, page: int) -> bytes:
return md5(base + struct.pack("<I", page) + b"sAlT")
def decrypt_page(base: bytes, data: bytes, page: int) -> bytes:
ct = data[(page - 1) * 4096:page * 4096]
for modulus in IV_MODULI:
pt = AES.new(page_key(base, page), AES.MODE_CBC,
page_iv(page, modulus)).decrypt(ct)
if pt[0] in BTREE_TYPES:
return pt
return pt # overflow/free page: default to the newest generation
def decrypt_file(src: str, dst: str, passphrase: str) -> None:
base = derive_base_key(passphrase)
data = open(src, "rb").read()
if len(data) % 4096:
raise SystemExit("file size is not a multiple of the 4096-byte page")
pages = len(data) // 4096
stream = data[8:16] + data[24:4096] # page 1: rejoin block 1 around header
page1 = (b"SQLite format 3\x00"
+ AES.new(page_key(base, 1), AES.MODE_CBC,
page_iv(1, IV_MODULI[0])).decrypt(stream))
out = bytearray(page1)
for p in range(2, pages + 1):
out += decrypt_page(base, data, p)
open(dst, "wb").write(out)
print(f"{src}: {pages} pages -> {dst}")
if __name__ == "__main__":
if len(sys.argv) != 4:
raise SystemExit(__doc__)
decrypt_file(sys.argv[1], sys.argv[2], sys.argv[3])
<!DOCTYPE html>
<html lang="th">
<head>
LINE Reader
:root {
--green: #06c755; --bg: #e6f5ec; --panel: #f7f7f7;
--line-dark: #1f2937; --line-gray: #9aa0a6;
}
* { box-sizing: border-box; margin: 0; padding: 0; }
body { font-family: 'Sarabun', 'LINESeedSansTH', -apple-system, 'Thonburi', sans-serif;
height: 100vh; display: flex; color: var(--line-dark); background: var(--bg); }
/* sidebar */
#side { width: 340px; min-width: 340px; background: #fff; border-right: 1px solid #e5e7eb;
display: flex; flex-direction: column; }
#brand { padding: 14px 16px; font-weight: 700; color: var(--green); font-size: 18px;
border-bottom: 1px solid #f0f0f0; display:flex; gap:8px; align-items:center; }
#search { padding: 10px 12px; border-bottom: 1px solid #f0f0f0; display: flex; gap: 8px; }
#search input { flex: 1; padding: 8px 12px; border: 1px solid #e5e7eb; border-radius: 18px;
font-size: 14px; outline: none; font-family: inherit; }
#search button { border: 0; background: var(--green); color: #fff; border-radius: 18px;
padding: 0 14px; cursor: pointer; font-family: inherit; }
#list { flex: 1; overflow-y: auto; }
.chat { display: flex; gap: 12px; padding: 12px 14px; cursor: pointer; align-items: center; }
.chat:hover { background: #f5f7f6; }
.chat.active { background: #eef9f1; }
.avatar { width: 46px; height: 46px; border-radius: 50%; color: #fff; font-weight: 700;
display: flex; align-items: center; justify-content: center; font-size: 18px; flex: none; }
.cmeta { flex: 1; min-width: 0; }
.crow { display: flex; justify-content: space-between; align-items: baseline; }
.cname { font-weight: 600; font-size: 15px; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
.ctime { font-size: 11px; color: var(--line-gray); flex: none; margin-left: 8px; }
.clast { font-size: 13px; color: #6b7280; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; margin-top: 2px; }
.badge { font-size: 11px; color: #9aa0a6; }
/* main */
#main { flex: 1; display: flex; flex-direction: column; min-width: 0; }
#head { background: #fff; border-bottom: 1px solid #e5e7eb; padding: 14px 20px;
font-weight: 700; font-size: 16px; display: flex; align-items: center; gap: 10px; }
#msgs { flex: 1; overflow-y: auto; padding: 18px 22px; display: flex; flex-direction: column; gap: 4px; }
.date { text-align: center; margin: 14px 0 8px; }
.date span { background: rgba(0,0,0,.35); color: #fff; font-size: 12px; padding: 3px 12px; border-radius: 12px; }
.mrow { display: flex; gap: 8px; margin-top: 6px; align-items: flex-end; }
.mrow.me { flex-direction: row-reverse; }
.bub { max-width: 62%; padding: 8px 12px; border-radius: 16px; font-size: 14.5px;
line-height: 1.45; white-space: pre-wrap; word-break: break-word; position: relative; }
.mrow:not(.me) .bub { background: #fff; border-bottom-left-radius: 4px; }
.mrow.me .bub { background: var(--green); color: #fff; border-bottom-right-radius: 4px; }
.mrow .who { font-size: 11.5px; color: #6b7280; margin-bottom: 2px; visibility: hidden; }
.grp .mrow:not(.me) .who { visibility: visible; }
.mrow .mt { font-size: 10.5px; color: var(--line-gray); align-self: flex-end; padding: 0 2px 3px; }
.sys { text-align: center; font-size: 12px; color: #6b7280; margin: 6px 0; }
.media, .sticker { font-size: 26px; padding: 6px 10px; }
#older { text-align: center; padding: 8px; }
#older button { border: 0; background: rgba(255,255,255,.8); border-radius: 14px;
padding: 5px 16px; cursor: pointer; font-size: 12.5px; color: #374151; font-family: inherit; }
.empty { margin: auto; color: #9aa0a6; font-size: 15px; }
#hits { max-height: 340px; overflow-y: auto; position: absolute; background: #fff; border: 1px solid #e5e7eb;
border-radius: 8px; width: calc(340px - 24px); box-shadow: 0 8px 24px rgba(0,0,0,.08); z-index: 5; }
.hit { padding: 8px 12px; cursor: pointer; border-bottom: 1px solid #f3f4f6; }
.hit:hover { background: #f5f7f6; }
.hit .hn { font-weight: 600; font-size: 13px; }
.hit .hx { font-size: 12px; color: #6b7280; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
</style>
</head>
<body>
💬 LINE Reader
</div>
</div>
เลือกแชทจากรายการด้านซ้าย
← เลือกแชทเพื่ออ่าน
</div>
let ME = '', chats = [], cur = null, oldest = null;
const COLORS = ['#06c755','#f7b500','#7c5cff','#ff6699','#00b8d9','#ff8c42','#31a24c','#e8505b'];
const colorFor = s => COLORS[[...(s||'x')].reduce((a,c)=>a+c.charCodeAt(0),0) % COLORS.length];
const initial = s => ((s||'?').replace(/^[ucrs]/,'')[0] || '?').toUpperCase();
const fmtTime = ms => { const d = new Date(ms); return d.getHours().toString().padStart(2,'0') + ':' + d.getMinutes().toString().padStart(2,'0'); };
const fmtDay = ms => { const d = new Date(ms); return d.toLocaleDateString('th-TH', {year:'numeric', month:'long', day:'numeric', weekday:'short'}); };
const sameDay = (a,b) => a && b && new Date(a).toDateString() === new Date(b).toDateString();
const esc = s => (s||'').replace(/[&<>"]/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));
async function loadChats() {
const r = await (await fetch('/api/chats')).json();
ME = r.me; chats = r.chats;
document.getElementById('count').textContent = chats.length + ' แชท';
renderList('');
}
function renderList(filter) {
const el = document.getElementById('list');
el.innerHTML = chats.filter(c => !filter || c.name.includes(filter)).map(c => `
${initial(c.name)}
${esc(c.name)}${c.last_t?fmtTime(c.last_t):''}
${esc(c.last_text)||'…'}${c.count} ข้อความ
</div>
</div>`).join('');
}
async function openChat(id) {
cur = id; renderList('');
const c = chats.find(x => x.id === id);
document.getElementById('head').innerHTML =
`${initial(c.name)}
${esc(c.name)} ${c.count} ข้อความ · ${c.kind}`;
const box = document.getElementById('msgs');
box.className = c.kind === 'dm' ? '' : 'grp';
box.innerHTML = 'กำลังโหลด…';
await loadMore(id, null, true);
}
async function loadMore(id, before, fresh) {
const r = await (await fetch('/api/messages?chat=' + encodeURIComponent(id) + (before ? '&before=' + before : ''))).json();
ME = r.me;
const box = document.getElementById('msgs');
const prevScroll = box.scrollHeight - box.scrollTop;
const html = r.messages.map((m, i, arr) => {
const prev = arr[i-1];
const dayChange = !prev || !sameDay(prev.t, m.t);
const mine = m.me !== null ? m.me : (ME ? m.from === ME : false);
let body = '';
if (m.kind === 'text') body = `${esc(m.text)}`;
else if (m.kind === 'sticker') body = `🎟️`;
else if (m.kind === 'media') body = `${m.ctype===1?'🖼️':m.ctype===2?'🎬':m.ctype===3?'🎧':'📎'}`;
else if (m.kind === 'location') body = `📍 ตำแหน่ง`;
else if (m.kind === 'system' || m.kind === 'contact') return `${esc(m.text) || '⚙️ ' + m.ctype}`;
return (dayChange ? `${fmtDay(m.t)}` : '') +
`${mine?'':`${esc(m.name)}`}${body}${fmtTime(m.t)}`;
}).join('');
if (fresh) {
box.innerHTML = (r.older ? `` : '') + html;
box.scrollTop = box.scrollHeight;
} else {
document.getElementById('older')?.remove();
box.insertAdjacentHTML('afterbegin', (r.older ? `` : '') + html);
box.scrollTop = box.scrollHeight - prevScroll;
}
if (r.messages.length) oldest = r.messages[0].t;
}
function older() { if (cur && oldest) loadMore(cur, oldest, false); }
async function doSearch() {
const q = document.getElementById('q').value.trim();
const box = document.getElementById('hits');
if (!q) { box.style.display = 'none'; return; }
const r = await (await fetch('/api/search?q=' + encodeURIComponent(q))).json();
box.innerHTML = r.hits.map(h => `
${esc(h.chat_name)} · ${new Date(h.t).toLocaleString('th-TH')}
${esc(h.from)}: ${esc(h.text)}`).join('') ||
'ไม่พบ';
box.style.display = 'block';
}
function jump(rowid, chat) {
document.getElementById('hits').style.display = 'none';
openChat(chat);
}
document.getElementById('q').addEventListener('keydown', e => { if (e.key === 'Enter') doSearch(); });
loadChats();
</script>
</body>
</html>
<!DOCTYPE html>
<html lang="th">
<head>
LINE Reader — LIVE
:root { --green: #06c755; --bg: #e6f5ec; --gray: #9aa0a6; }
* { box-sizing: border-box; margin: 0; padding: 0; }
body { font-family: 'Sarabun', -apple-system, 'Thonburi', sans-serif; height: 100vh;
display: flex; color: #1f2937; background: var(--bg); }
#side { width: 340px; min-width: 340px; background: #fff; border-right: 1px solid #e5e7eb; display: flex; flex-direction: column; }
#brand { padding: 12px 16px; font-weight: 700; color: var(--green); font-size: 18px;
border-bottom: 1px solid #f0f0f0; display: flex; align-items: center; gap: 8px; }
#live { font-size: 11px; background: #e5e7eb; color: #6b7280; border-radius: 10px; padding: 2px 10px; font-weight: 600; }
#live.on { background: var(--green); color: #fff; animation: pulse 2s infinite; }
@keyframes pulse { 0%,100% { opacity: 1 } 50% { opacity: .55 } }
#status { font-size: 11px; color: var(--gray); margin-left: auto; font-weight: 400; text-align: right; }
#bell { border: 0; background: none; cursor: pointer; font-size: 17px; }
#bell.off { filter: grayscale(1); opacity: .45; }
#search { padding: 10px 12px; border-bottom: 1px solid #f0f0f0; }
#search input { width: 100%; padding: 8px 12px; border: 1px solid #e5e7eb; border-radius: 18px; font-size: 14px; outline: none; font-family: inherit; }
#list { flex: 1; overflow-y: auto; }
.chat { display: flex; gap: 12px; padding: 12px 14px; cursor: pointer; align-items: center; }
.chat:hover { background: #f5f7f6; } .chat.active { background: #eef9f1; }
.avatar { width: 46px; height: 46px; border-radius: 50%; color: #fff; font-weight: 700;
display: flex; align-items: center; justify-content: center; font-size: 18px; flex: none; position: relative; }
.dot { position: absolute; top: -2px; right: -4px; background: #ef4444; color: #fff; font-size: 10px;
min-width: 17px; height: 17px; border-radius: 9px; display: flex; align-items: center;
justify-content: center; padding: 0 4px; font-weight: 700; }
.cmeta { flex: 1; min-width: 0; }
.crow { display: flex; justify-content: space-between; align-items: baseline; }
.cname { font-weight: 600; font-size: 15px; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
.ctime { font-size: 11px; color: var(--gray); margin-left: 8px; }
.clast { font-size: 13px; color: #6b7280; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; margin-top: 2px; }
.badge { font-size: 11px; color: var(--gray); }
#main { flex: 1; display: flex; flex-direction: column; min-width: 0; }
#head { background: #fff; border-bottom: 1px solid #e5e7eb; padding: 14px 20px; font-weight: 700; font-size: 16px; }
#msgs { flex: 1; overflow-y: auto; padding: 18px 22px; }
.date { text-align: center; margin: 14px 0 8px; }
.date span { background: rgba(0,0,0,.35); color: #fff; font-size: 12px; padding: 3px 12px; border-radius: 12px; }
.mrow { display: flex; gap: 8px; margin-top: 6px; align-items: flex-end; }
.mrow.me { flex-direction: row-reverse; }
.bub { max-width: 62%; padding: 8px 12px; border-radius: 16px; font-size: 14.5px; line-height: 1.45;
white-space: pre-wrap; word-break: break-word; }
.mrow:not(.me) .bub { background: #fff; border-bottom-left-radius: 4px; }
.mrow.me .bub { background: var(--green); color: #fff; border-bottom-right-radius: 4px; }
.who { font-size: 11.5px; color: #6b7280; margin-bottom: 2px; }
.mt { font-size: 10.5px; color: var(--gray); align-self: flex-end; padding: 0 2px 3px; }
.sys { text-align: center; font-size: 12px; color: #6b7280; margin: 6px 0; }
.media, .sticker { font-size: 26px; padding: 6px 10px; }
.fresh .bub { box-shadow: 0 0 0 2px rgba(6,199,85,.45); }
.empty { margin: auto; color: var(--gray); font-size: 15px; text-align: center; line-height: 2; }
#toasts { position: fixed; right: 16px; top: 16px; z-index: 50; display: flex; flex-direction: column; gap: 8px; }
.toast { background: #fff; border-left: 4px solid var(--green); border-radius: 8px;
box-shadow: 0 8px 24px rgba(0,0,0,.12); padding: 10px 14px; min-width: 240px; max-width: 320px;
cursor: pointer; animation: slide .25s; }
@keyframes slide { from { transform: translateX(30px); opacity: 0 } }
.toast .tn { font-weight: 700; font-size: 13.5px; }
.toast .tx { font-size: 12.5px; color: #6b7280; margin-top: 2px; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
</style>
</head>
<body>
💬 LINE Reader ⏳
…
</div>
เลือกแชทจากรายการด้านซ้าย
← เลือกแชทเพื่ออ่าน
โหมด LIVE: ข้อความใหม่จะเด้งขึ้นเอง
</div>
let ME = '', chats = [], cur = null, version = 0, unread = {}, bell = true;
const COLORS = ['#06c755','#f7b500','#7c5cff','#ff6699','#00b8d9','#ff8c42','#31a24c','#e8505b'];
const colorFor = s => COLORS[[...(s||'x')].reduce((a,c)=>a+c.charCodeAt(0),0) % COLORS.length];
const initial = s => ((s||'?').replace(/^[ucrs]/,'')[0] || '?').toUpperCase();
const fmtTime = ms => { const d = new Date(ms); return String(d.getHours()).padStart(2,'0')+':'+String(d.getMinutes()).padStart(2,'0'); };
const fmtDay = ms => new Date(ms).toLocaleDateString('th-TH',{year:'numeric',month:'long',day:'numeric',weekday:'short'});
const sameDay = (a,b) => a && b && new Date(a).toDateString() === new Date(b).toDateString();
const esc = s => (s||'').replace(/[&<>"]/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));
const $ = s => document.querySelector(s);
async function loadChats() {
const r = await (await fetch('/api/chats')).json();
ME = r.me; version = Math.max(version, r.version); chats = r.chats;
$('#status').textContent = r.status || '';
$('#live').textContent = 'LIVE'; $('#live').classList.add('on');
renderList($('#q').value.trim());
}
function renderList(filter) {
$('#list').innerHTML = chats
.filter(c => !filter || c.name.includes(filter) || c.last_text.includes(filter))
.map(c => `
${initial(c.name)}
${unread[c.id] ? `${unread[c.id]}` : ''}
${esc(c.name)}${c.last_t?fmtTime(c.last_t):''}
${esc(c.last_text)||'…'}${c.count}
</div></div>`).join('');
}
async function openChat(id) {
cur = id; delete unread[id]; renderList($('#q').value.trim());
const c = chats.find(x => x.id === id);
$('#head').innerHTML = `${initial(c.name)}${esc(c.name)} ${c.count} ข้อความ · ${c.kind}`;
await loadMsgs(id, null, false);
}
async function loadMsgs(id, before, prepend) {
const r = await (await fetch('/api/messages?chat=' + encodeURIComponent(id) + (before ? '&before=' + before : ''))).fetch
? {} : await (await fetch('/api/messages?chat=' + encodeURIComponent(id) + (before ? '&before=' + before : ''))).json();
const box = $('#msgs');
box.className = id[0] === 'u' ? '' : 'grp';
const html = r.messages.map((m, i) => {
const prev = r.messages[i-1];
const dayChange = !prev || !sameDay(prev.t, m.t);
let body;
if (m.kind === 'text') body = `${esc(m.text)}`;
else if (m.kind === 'sticker') body = `🎟️`;
else if (m.kind === 'media') body = `${m.ctype===1?'🖼️':m.ctype===2?'🎬':m.ctype===3?'🎧':'📎'}`;
else if (m.kind === 'location') body = `📍 ตำแหน่ง`;
else return (dayChange ? `${fmtDay(m.t)}` : '') + `${esc(m.text)||'⚙️'}`;
return (dayChange ? `${fmtDay(m.t)}` : '') +
`${m.me?'':`${esc(m.name)}`}${body}${fmtTime(m.t)}`;
}).join('');
if (prepend) box.insertAdjacentHTML('beforeend', html);
else { box.innerHTML = html; box.scrollTop = box.scrollHeight; }
}
function toast(name, text, chat) {
const el = document.createElement('div');
el.className = 'toast';
el.innerHTML = `💬 ${esc(name)}${esc(text)}`;
el.onclick = () => { openChat(chat); el.remove(); };
$('#toasts').appendChild(el);
setTimeout(() => el.remove(), 8000);
}
function notify(name, text, chat) {
toast(name, text, chat);
if (!bell) return;
if ('Notification' in window && Notification.permission === 'granted') {
try {
const n = new Notification(name, { body: text.slice(0, 120), tag: chat });
n.onclick = () => { window.focus(); openChat(chat); n.close(); };
} catch (e) {}
}
}
async function toggleBell() {
bell = !bell;
$('#bell').classList.toggle('off', !bell);
if (bell && 'Notification' in window && Notification.permission === 'default')
Notification.requestPermission();
}
async function pollLoop() {
for (;;) {
try {
const r = await (await fetch('/api/poll?since=' + version)).json();
$('#status').textContent = (r.status || '').slice(0, 26);
if (!r.timeout && r.version > version) {
const known = version;
version = r.version;
const changes = r.changes || {};
await loadChats();
for (const [cid, ch] of Object.entries(changes)) {
if (cid === cur) {
loadMsgs(cid, null, false);
if (ch.last_text) notify(ch.name, ch.last_from + ': ' + ch.last_text, cid);
} else {
unread[cid] = (unread[cid] || 0) + ch.new;
if (ch.last_text) notify(ch.name, ch.last_from + ': ' + ch.last_text, cid);
}
}
renderList($('#q').value.trim());
document.title = Object.keys(unread).length
? `(${Object.values(unread).reduce((a,b)=>a+b,0)}) LINE Reader — LIVE` : 'LINE Reader — LIVE';
}
} catch (e) { await new Promise(r => setTimeout(r, 3000)); }
}
}
if ('Notification' in window && Notification.permission === 'default')
Notification.requestPermission();
loadChats().then(pollLoop);
</script>
</body>
</html>
#!/usr/bin/env python3
"""LINE chat reader — read-only web UI over the decrypted SQLite DB.
Loads all readable rows into memory at startup (skips corrupt page windows),
then serves everything from RAM. Usage: server.py [db.sqlite] [port]
"""
import json
import os
import sqlite3
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
ROOT = os.path.dirname(os.path.abspath(__file__))
DB = os.path.abspath(os.path.expanduser(sys.argv[1] if len(sys.argv) > 2
else '~/tmp/line-decrypted/main.sqlite'))
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8819
con = sqlite3.connect(DB)
con.row_factory = sqlite3.Row
def safe(fn):
try:
return fn()
except Exception:
return []
def scan_table(sql, cols, maxrow=4000, chunk=100):
"""Rowid-chunked scan that skips corrupt page windows."""
out = []
for start in range(0, maxrow, chunk):
rows = safe(lambda: con.execute(
sql + ' WHERE rowid>=? AND rowid<? LIMIT ?', (start, start + chunk, chunk)).fetchall())
out.extend(tuple(r[c] for c in cols) for r in rows)
return out
# ---- static data ----
import json as _json
CONTACTS = {}
for _mid, _det, _over, _phone in scan_table(
'SELECT rowid, _mid, _targetProfileDetail, _displayNameOverridden, _phoneDisplayName FROM _contact',
('_mid', '_targetProfileDetail', '_displayNameOverridden', '_phoneDisplayName')):
_name = _over or _phone
if not _name and _det:
try:
_name = _json.loads(_det).get('profileName')
except Exception:
_name = None
if _name:
CONTACTS[_mid] = _name
ME = ''
for col in ('_mid', 'mid', '_midOrId'):
rows = safe(lambda: con.execute('SELECT %s FROM _profile LIMIT 1' % col).fetchall())
if rows and rows[0][0]:
ME = rows[0][0]
break
# ---- messages: chunked rowid scan, skipping corrupt windows ----
MSGS = [] # list of dicts, ordered by rowid
MAXROW = 12000
CHUNK = 200
bad_blocks = 0
for start in range(0, MAXROW, CHUNK):
try:
rows = con.execute(
'SELECT rowid,_chatId,_createdTime,_from,_to,_toType,_text,_contentType '
'FROM _message WHERE rowid>=? AND rowid<? ORDER BY rowid LIMIT ?',
(start, start + CHUNK, CHUNK)).fetchall()
except Exception:
bad_blocks += 1
continue
for r in rows:
MSGS.append({
'rowid': r['rowid'], 'chat': r['_chatId'], 't': r['_createdTime'] or 0,
'from': r['_from'] or '', 'to': r['_to'] or '',
'toType': r['_toType'], 'text': r['_text'] or '',
'ctype': r['_contentType'] or 0,
})
MSGS.sort(key=lambda m: m['t'])
# ---- derived: chat list ----
CHATS = {}
for m in MSGS:
c = CHATS.setdefault(m['chat'], {'id': m['chat'], 'n': 0, 'last_t': 0, 'last_text': ''})
c['n'] += 1
if m['t'] >= c['last_t']:
c['last_t'] = m['t']
c['last_text'] = m['text']
def chat_name(cid):
return GROUPS.get(cid) or CONTACTS.get(cid) or cid
CHATS = sorted(CHATS.values(), key=lambda c: c['last_t'], reverse=True)
print('DB=%s | me=%s | msgs=%d (bad blocks=%d) | chats=%d'
% (DB, ME or '?', len(MSGS), bad_blocks, len(CHATS)), flush=True)
def classify(m):
t, text = m['ctype'], m['text']
if t == 0 and not text:
return 'system'
if t in (1, 2, 3, 4):
return 'media'
if t == 7:
return 'location'
if t == 8:
return 'sticker'
if t == 9:
return 'contact'
if text.startswith('{'):
return 'system'
return 'text'
def api_chats(q):
return {'me': ME, 'chats': [{
'id': c['id'], 'name': chat_name(c['id']),
'count': c['n'], 'last_t': c['last_t'], 'last_text': c['last_text'][:80],
'kind': 'group' if c['id'].startswith('c') else ('room' if c['id'].startswith('r') else 'dm'),
} for c in CHATS]}
def api_messages(q):
chat = q.get('chat', [''])[0]
before = int(q['before'][0]) if q.get('before') else None
sel = [m for m in MSGS if m['chat'] == chat and (before is None or m['t'] < before)]
sel = sel[-100:]
out = []
for m in sel:
kind = classify(m)
out.append({
'rowid': m['rowid'], 't': m['t'],
'from': m['from'], 'name': CONTACTS.get(m['from'], m['from'] or ''),
'me': (m['from'] == ME) if ME else False,
'text': m['text'] if kind == 'text' else '', 'kind': kind, 'ctype': m['ctype'],
})
return {'messages': out, 'older': len([m for m in MSGS if m['chat'] == chat]) > 0
and sel and sel[0]['t'] > min(m['t'] for m in MSGS if m['chat'] == chat),
'me': ME}
def api_search(q):
term = q.get('q', [''])[0].strip()
if not term:
return {'hits': []}
hits = []
for m in reversed(MSGS):
if term in m['text']:
hits.append({
'rowid': m['rowid'], 'chat': m['chat'], 'chat_name': chat_name(m['chat']),
't': m['t'], 'from': CONTACTS.get(m['from'], m['from'] or ''),
'text': m['text'][:160],
})
if len(hits) >= 50:
break
return {'hits': hits}
class H(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def _send(self, code, body, ctype='application/json; charset=utf-8'):
if isinstance(body, str):
body = body.encode()
self.send_response(code)
self.send_header('Content-Type', ctype)
self.send_header('Content-Length', str(len(body)))
self.send_header('Cache-Control', 'no-store')
self.end_headers()
self.wfile.write(body)
def do_GET(self):
u = urlparse(self.path)
q = parse_qs(u.query)
try:
if u.path == '/api/chats':
return self._send(200, json.dumps(api_chats(q), ensure_ascii=False))
if u.path == '/api/messages':
return self._send(200, json.dumps(api_messages(q), ensure_ascii=False))
if u.path == '/api/search':
return self._send(200, json.dumps(api_search(q), ensure_ascii=False))
path = u.path.lstrip('/') or 'index.html'
fn = os.path.join(ROOT, path)
if not os.path.isfile(fn):
return self._send(404, 'not found', 'text/plain')
ctype = ('text/html; charset=utf-8' if fn.endswith('.html') else
'text/css' if fn.endswith('.css') else 'application/javascript')
self._send(200, open(fn, 'rb').read(), ctype)
except Exception as e:
self._send(500, json.dumps({'error': str(e)}))
if __name__ == '__main__':
print('serving on http://127.0.0.1:%d' % PORT, flush=True)
ThreadingHTTPServer(('127.0.0.1', PORT), H).serve_forever()
#!/usr/bin/env python3
"""LINE chat reader — live edition.
Watches the *encrypted* .edb + -wal of the running LINE app, re-decrypts on
change (per-page AES-128-CBC + WAL frame overlay), and serves an auto-updating
web UI. Usage: server.py [decrypted-bootstrap.sqlite] [port]
The bootstrap DB (from decrypt_line.py) provides full history; live updates
merge the running app's WAL frames on top.
"""
import json
import os
import sqlite3
import struct
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
from Crypto.Cipher import AES
from Crypto.Hash import MD5
ROOT = os.path.dirname(os.path.abspath(__file__))
BOOT = os.path.abspath(os.path.expanduser(sys.argv[1] if len(sys.argv) > 1
else '~/tmp/line-decrypted/main.sqlite'))
EDB = os.path.expanduser(
'~/Library/Containers/jp.naver.line.mac/Data/Library/Containers/'
'jp.naver.line/Data/db/qwa31d6c43cda620de64b950a4f6bc75.edb')
WAL = EDB + '-wal'
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8819
PASSPHRASE = '4d7941d0d0e241900050db5e451f1299'
BASE = None # derived below
BTREE = (2, 5, 10, 13)
IV_MODULI = (0x7FFFFFC7, 0x7FFFFF07)
def md5(b):
return MD5.new(b).digest()
def pad(pw):
P = bytes.fromhex('28bf4e5e4e758a4164004e56fffa0108'
'2e2e00b6d0683e802f0ca9fe6453697a')
pb = pw.encode()
return pb[:32] if len(pb) >= 32 else pb + P[len(pb):]
def rc4(key, data):
s = list(range(256))
j = 0
for i in range(256):
j = (j + s[i] + key[i % len(key)]) & 255
s[i], s[j] = s[j], s[i]
i = j = 0
out = bytearray()
for b in data:
i = (i + 1) & 255
j = (j + s[i]) & 255
s[i], s[j] = s[j], s[i]
out.append(b ^ s[(s[i] + s[j]) & 255])
return bytes(out)
def derive_base(pw):
up, op = pad(pw), pad('')
d = md5(op)
for _ in range(50):
d = md5(d)
ok = bytearray(up)
for r in range(20):
ok = bytearray(rc4(bytes(x ^ r for x in d), bytes(ok)))
d = md5(up + bytes(ok))
for _ in range(50):
d = md5(d)
return d[:16]
BASE = derive_base(PASSPHRASE)
def lcg(z, m):
q = z // 52774
z = 40692 * (z - 52774 * q) - 3791 * q
return z + m if z < 0 else z
def page_iv(p, m):
z = p + 1
k = bytearray(16)
for j in range(4):
z = lcg(z, m)
k[4 * j:4 * j + 4] = struct.pack('<I', z)
return md5(bytes(k))
def page_key(p):
return md5(BASE + struct.pack('<I', p) + b'sAlT')
def dec_page(ct, p):
outs = []
for m in IV_MODULI:
pt = AES.new(page_key(p), AES.MODE_CBC, page_iv(p, m)).decrypt(ct)
if pt[0] in BTREE:
return pt
outs.append(pt)
return outs[0]
def dec_page1(data):
stream = data[8:16] + data[24:4096]
return b'SQLite format 3\x00' + AES.new(
page_key(1), AES.MODE_CBC, page_iv(1, IV_MODULI[0])).decrypt(stream)
# ---------- live state ----------
STATE = {'version': 0, 'updated': 0, 'msg': 'starting', 'source': ''}
LOCK = threading.Lock()
MSGS = [] # dicts: rowid, chat, t, from, text, ctype
CONTACTS = {}
GROUPS = {}
ME = ''
DBFILE = '/tmp/line-live.sqlite'
def safe(con, fn):
try:
return fn()
except Exception:
return []
def scan(con, sql, cols, maxrow, chunk=200):
out = []
for s in range(0, maxrow, chunk):
rows = safe(con, lambda: con.execute(
sql + ' WHERE rowid>=? AND rowid<? LIMIT ?', (s, s + chunk, chunk)).fetchall())
out.extend(tuple(r[c] for c in cols) for r in rows)
return out
def scan_deep(con, sql, cols, maxrow, chunk=500):
"""Chunked scan that bisects corrupt-page windows down to single rows."""
out = []
def rec(lo, hi, depth=0):
try:
rows = con.execute(sql + ' WHERE rowid>=? AND rowid<? ORDER BY rowid',
(lo, hi)).fetchall()
return rows
except Exception:
if hi - lo <= 1 or depth > 12:
return []
mid = (lo + hi) // 2
return rec(lo, mid, depth + 1) + rec(mid, hi, depth + 1)
for s in range(0, maxrow, chunk):
out.extend(rec(s, s + chunk))
return [tuple(r[c] for c in cols) for r in out]
def build_live_image():
"""Decrypt the running app's edb pages, overlay newest WAL frames."""
data = open(EDB, 'rb').read()
pages = {}
n = len(data) // 4096
for p in range(2, n + 1):
pages[p] = dec_page(data[(p - 1) * 4096:p * 4096], p)
pages[1] = dec_page1(data)
newest = {}
if os.path.exists(WAL):
wal = open(WAL, 'rb').read()
if len(wal) >= 32 and wal[:4] in (b'\x37\x7f\x06\x82', b'\x37\x7f\x06\x83'):
psz = struct.unpack('>I', wal[8:12])[0]
off = 32
while off + 24 + psz <= len(wal):
pgno, dbsize = struct.unpack('>II', wal[off:off + 8])
if pgno:
newest[pgno] = wal[off + 24:off + 24 + psz]
off += 24 + psz
for pgno, ct in newest.items():
pages[pgno] = dec_page1(ct) if pgno == 1 else dec_page(ct, pgno)
maxp = max(pages)
out = bytearray()
for p in range(1, maxp + 1):
out += pages.get(p, b'\x00' * 4096)
tmp = DBFILE + '.tmp'
open(tmp, 'wb').write(out)
os.replace(tmp, DBFILE)
return len(out) // 4096, len(newest)
def reload_from_db(path, source):
global ME
con = sqlite3.connect(path)
con.row_factory = sqlite3.Row
import json as _j
for mid, det, over, phone in scan_deep(con,
'SELECT rowid,_mid,_targetProfileDetail,_displayNameOverridden,_phoneDisplayName FROM _contact',
('_mid', '_targetProfileDetail', '_displayNameOverridden', '_phoneDisplayName'), 20000, 500):
name = over or phone
if not name and det:
try:
name = _j.loads(det).get('profileName')
except Exception:
name = None
if name:
CONTACTS[mid] = name
for mid, name in scan_deep(con, 'SELECT rowid,_chatMid,_chatName FROM _groupChat',
('_chatMid', '_chatName'), 4000, 500):
if name:
GROUPS[mid] = name
for col in ('_mid', 'mid', '_midOrId'):
rows = safe(con, lambda: con.execute('SELECT %s FROM _profile LIMIT 1' % col).fetchall())
if rows and rows[0][0]:
ME = rows[0][0]
break
msgs = []
for rowid, chat, t, frm, text, ctype in scan(con,
'SELECT rowid,_chatId,_createdTime,_from,_text,_contentType FROM _message',
('rowid', '_chatId', '_createdTime', '_from', '_text', '_contentType'), 20000):
msgs.append({'rowid': rowid, 'chat': chat, 't': t or 0, 'from': frm or '',
'text': text or '', 'ctype': ctype or 0})
msgs.sort(key=lambda m: m['t'])
con.close()
# diff per-chat message counts for notifications
prev = dict(STATE['counts']) if STATE.get('counts') else {}
cur_counts = {}
last_texts = {}
for m in msgs:
cur_counts[m['chat']] = cur_counts.get(m['chat'], 0) + 1
if m['t'] >= (last_texts.get(m['chat']) or (0,))[0]:
last_texts[m['chat']] = (m['t'], m['from'], m['text'])
changes = {c: cur_counts[c] - prev.get(c, 0) for c in cur_counts
if cur_counts[c] > prev.get(c, 0)}
with LOCK:
MSGS.clear()
MSGS.extend(msgs)
STATE['version'] += 1
STATE['updated'] = time.time()
STATE['source'] = source
STATE['counts'] = cur_counts
if changes:
STATE['changes'] = {
c: {'new': n,
'name': GROUPS.get(c) or CONTACTS.get(c) or c,
'last_from': CONTACTS.get(last_texts[c][1], last_texts[c][1] or ''),
'last_text': last_texts[c][2][:120],
'last_t': last_texts[c][0]}
for c, n in changes.items()}
STATE.setdefault('events', []).append(
{'t': time.time(), 'version': STATE['version'], 'changes': STATE['changes']})
STATE['events'] = STATE['events'][-100:]
else:
STATE['changes'] = {}
def watcher():
# bootstrap: history from the offline decrypt, then go live
if os.path.exists(BOOT):
try:
reload_from_db(BOOT, 'bootstrap')
STATE['msg'] = 'history loaded'
except Exception as e:
STATE['msg'] = 'bootstrap failed: %s' % e
last_sig = None
while True:
try:
sig = (os.path.getmtime(EDB) if os.path.exists(EDB) else 0,
os.path.getsize(EDB) if os.path.exists(EDB) else 0,
os.path.getmtime(WAL) if os.path.exists(WAL) else 0,
os.path.getsize(WAL) if os.path.exists(WAL) else 0)
if sig != last_sig and sig[1]:
npages, nframes = build_live_image()
reload_from_db(DBFILE, 'live (wal frames: %d)' % nframes)
STATE['msg'] = 'live · %d pages · wal %d frames' % (npages, nframes)
last_sig = sig
except Exception as e:
STATE['msg'] = 'watch error: %s' % e
time.sleep(2)
threading.Thread(target=watcher, daemon=True).start()
def classify(m):
t, text = m['ctype'], m['text']
if (t == 0 and not text) or text.startswith('{'):
return 'system'
if t in (1, 2, 3, 4):
return 'media'
if t == 7:
return 'location'
if t == 8:
return 'sticker'
return 'text'
def api_chats(q):
with LOCK:
by = {}
for m in MSGS: # MSGS sorted by t; rowid reflects DB write order
c = by.setdefault(m['chat'], {'id': m['chat'], 'n': 0, 'last_t': 0,
'last_text': '', 'last_rowid': 0})
c['n'] += 1
if m['rowid'] >= c['last_rowid']:
c['last_rowid'] = m['rowid']
c['last_t'] = m['t']
c['last_text'] = m['text']
chats = sorted(by.values(), key=lambda c: c['last_rowid'], reverse=True)
return {'me': ME, 'version': STATE['version'], 'status': STATE['msg'],
'chats': [{
'id': c['id'],
'name': GROUPS.get(c['id']) or CONTACTS.get(c['id']) or c['id'],
'count': c['n'], 'last_t': c['last_t'], 'last_text': c['last_text'][:80],
'kind': 'group' if c['id'].startswith('c') else ('room' if c['id'].startswith('r') else 'dm'),
} for c in chats]}
def api_messages(q):
chat = q.get('chat', [''])[0]
before = int(q['before'][0]) if q.get('before') else None
with LOCK:
sel = [m for m in MSGS if m['chat'] == chat and (before is None or m['t'] < before)]
sel = sel[-100:]
all_n = len([m for m in MSGS if m['chat'] == chat])
out = [{
'rowid': m['rowid'], 't': m['t'],
'from': m['from'], 'name': CONTACTS.get(m['from'], m['from'] or ''),
'me': m['from'] == ME if ME else False,
'text': m['text'] if classify(m) == 'text' else '',
'kind': classify(m), 'ctype': m['ctype'],
} for m in sel]
return {'messages': out, 'total': all_n, 'version': STATE['version'], 'me': ME,
'older': all_n > len(sel) and sel and all_n > 100}
def api_poll(q):
since = int(q.get('since', ['0'])[0])
deadline = time.time() + 25
while time.time() < deadline:
with LOCK:
if STATE['version'] > since:
return {'version': STATE['version'], 'updated': STATE['updated'],
'status': STATE['msg'], 'changes': STATE.get('changes', {})}
time.sleep(1)
with LOCK:
return {'version': STATE['version'], 'updated': STATE['updated'],
'status': STATE['msg'], 'timeout': True, 'changes': {}}
def api_search(q):
term = q.get('q', [''])[0].strip()
with LOCK:
hits = []
for m in reversed(MSGS):
if term and term in m['text']:
hits.append({'rowid': m['rowid'], 'chat': m['chat'],
'chat_name': GROUPS.get(m['chat']) or CONTACTS.get(m['chat']) or m['chat'],
't': m['t'], 'from': CONTACTS.get(m['from'], m['from'] or ''),
'text': m['text'][:160]})
if len(hits) >= 50:
break
return {'hits': hits}
class H(BaseHTTPRequestHandler):
def log_message(self, *a):
pass
def _send(self, code, body, ctype='application/json; charset=utf-8'):
if isinstance(body, str):
body = body.encode()
self.send_response(code)
self.send_header('Content-Type', ctype)
self.send_header('Content-Length', str(len(body)))
self.send_header('Cache-Control', 'no-store')
self.end_headers()
self.wfile.write(body)
def do_GET(self):
u = urlparse(self.path)
q = parse_qs(u.query)
try:
if u.path == '/api/chats':
return self._send(200, json.dumps(api_chats(q), ensure_ascii=False))
if u.path == '/api/messages':
return self._send(200, json.dumps(api_messages(q), ensure_ascii=False))
if u.path == '/api/search':
return self._send(200, json.dumps(api_search(q), ensure_ascii=False))
if u.path == '/api/poll':
return self._send(200,
添加评论
点赞收藏
点踩分享查看原文
评论
?
参与讨论