本文代码的原理基于 git@github.com:mcxiaoke/packer-ng-plugin.git 项目。

该项目用于向打好的 APK 包快速写入渠道信息，因为是直接在 APK 的尾巴上加数据而没有解包打包的过程，故速度较快。本文实现的是另一种需求，即后台实时给 APK 加数据，然后返回给前端下载。

比如邀请码这样没法提前写入的数据，就适合实时生成。得到的好处就是用户在注册时可以免去填邀请码这一步。

但显然这会带来带宽的压力，依旧去实现它是考虑到了实际分享应用的场景：绝大多数是在 微信、QQ 等平台内传播，都会跳转到应用宝等市场上去。因此这种实时打包流量其实很低，除非有人非要别扭着在微信里使用【在浏览器中打开】按钮来下载，否则本接口是不会被访问到的。

所以这个实现其实一点实际用处都没有 →_→

原理

APK 使用的是 ZIP 的打包格式，所以它的结尾定义了两个字段 —— Comment Len, Comment，Comment 是个变长字段，长度写在 Comment Len 里，占两个字节。

ZIP 的详细解释参见：

https://users.cs.jmu.edu/buchhofp/forensics/formats/pkzip.html

APK 在安装后，还会在系统里留下一个副本，安装过的应用通过读取这个副本的 Comment，就可以获得这个数据。

PackerNg

开头的 github 项目便实现了这个功能，通过 Java 和 Python 两种语言。本文重写了 Python 的版本，使其可以更改 Comment，以便实现动态加尾巴的功能。

PackerNg 的 comment 格式 是 data + data length(2 bytes) + !ZXK!。最后的 !ZXK! 是个 magic 字段，用于判断该文件是否有有效的尾巴，再通过这个字段之前的两个字节来获取偏移量，读取尾巴。

但我感觉这个字段是可选的，如果直接以偏移量结尾，我们便可以通过检查最后两个字节是不是 "\x00\x00" 来判断该文件有没有加尾巴。可能的混淆情景仅为，该文件有一个非我们加的尾巴，而这可以通过人为约束来控制。

加尾巴代码

APK_COMM_LEN = 2  # bytes
APK_PACK_FMT = '<H'  # little endian unsigned short
APK_MAGIC_WD = b'!ZXK!'
APK_MAGIC_LEN = 5
# stream 相关
APK_BASE_CHUNK_SIZE = 1024 * 1024
APK_COMM_SAFE_LEN = 1024  # 保证最后一个 apk chunk 的尾部有这么多字节# ------------------------ APK ------------------------ #
def get_apk_comment(f):"""获取 apk 包尾部的 comment 字节, 含 MAGIC"""# read magicf.seek(-APK_MAGIC_LEN, 2)tail = f.read(APK_MAGIC_LEN)if tail[-APK_COMM_LEN:] == b'\x00\x00':return b''if tail != APK_MAGIC_WD:raise ValueError("invalid apk comment format, it's not ended with MAGIC.")# read comment lenf.seek(-APK_MAGIC_LEN - APK_COMM_LEN, 2)comm_len = struct.unpack(APK_PACK_FMT, f.read(APK_COMM_LEN))[0]f.seek(-APK_MAGIC_LEN - APK_COMM_LEN - comm_len, 2)# read commentcomment = f.read(comm_len + APK_COMM_LEN + APK_MAGIC_LEN)return commentdef get_apk_comm_info(f):"""获取 apk 包尾部的 json 数据"""comment = get_apk_comment(f)if not comment:return {}comm_info = comment[:-APK_COMM_LEN - APK_MAGIC_LEN]if comm_info:return json.loads(comm_info.decode())def gen_apk_comment(info={}):"""通过字典生成 comm_len 和 comment，均为 bytes 类型，含 MAGIC"""if not info:return b'\x00\x00', b''comment = json.dumps(info).encode()comment += struct.pack('<H', len(comment))comment += APK_MAGIC_WDreturn struct.pack('<H', len(comment)), commentdef set_apk_comment(f, cur_comm_len=b'\x00\x00', new_comm_len=b'\x00\x00', new_comment=b''):"""设置 apk 包尾部的 comment 字节，全量更新"""# check cur_comm_lencur_comm_len = struct.unpack(APK_PACK_FMT, cur_comm_len)[0]if cur_comm_len == 0:f.seek(-APK_COMM_LEN, 2)if f.read(APK_COMM_LEN) != b'\x00\x00':cur_comm = get_apk_comment(f)cur_comm_len = len(cur_comm)# write newf.seek(-cur_comm_len - APK_COMM_LEN, 2)f.write(new_comm_len)f.write(new_comment)f.truncate()f.seek(0, 0)return f

流式加尾巴代码

因为 Comment 信息位于 APK 的尾部，而我司渠道包又非常多且迭代较快，我没有选择在服务器本地保存 APK 包，而是实时从 cdn 取数据流，然后在最后一段上进行更改，即实现成了一个代理。

def inv_apk_wrapper(url, inv_code):r = requests.get(url, timeout=300, stream=True)# 生成符合要求的 chunk_sizecontent_length = json.loads(r.headers['content-length'])tail_length = content_length % APK_BASE_CHUNK_SIZEif tail_length < APK_COMM_SAFE_LEN:less = APK_COMM_SAFE_LEN - tail_lengthchunk_num = int(math.ceil(content_length / float(APK_BASE_CHUNK_SIZE)))chunk_size = APK_BASE_CHUNK_SIZE - int(math.ceil(less / (chunk_num - 1.0)))else:chunk_size = APK_BASE_CHUNK_SIZEnew_apk_gen = inv_apk_generator(r.iter_content(chunk_size), inv_code)# update content-lengthnew_headers = dict(r.headers)new_headers['content-length'] = str(content_length + len(', "inv_code": "{inv_code}"'.format(inv_code=inv_code)))return Response(new_apk_gen, headers=new_headers)def inv_apk_generator(apk_iterator, inv_code):data = []while True:try:data.append(next(apk_iterator))except StopIteration:  # 确认到尾部, 添加 inv_codeif data:content = BytesIO(data[0])comm_info = get_apk_comm_info(content)comm_info['inv_code'] = inv_codecomm_len, comment = gen_apk_comment(comm_info)content = set_apk_comment(content, new_comm_len=comm_len, new_comment=comment)content.seek(0, 0)yield content.read()raise StopIterationif len(data) >= 2:yield data[0]data = data[1:]

注意这种接口因为太占 I/O，最好不要和正常业务部署在一起，或者，最好不要部署。。。