-
-
Save muayyad-alsadi/2851a002b3ce1c53de763145603d7f4f to your computer and use it in GitHub Desktop.
grpc-web reverse engineer, just pass the minified javascript file
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
syntax = "proto3"; | |
package helloworld; | |
service Greeter { | |
rpc SayHello (HelloRequest) returns (HelloResponse); | |
rpc SearchResults (Empty) returns (SearchResponse); | |
} | |
message HelloRequest { | |
string name = 1; | |
} | |
message HelloResponse { | |
string message = 1; | |
string test = 2; | |
} | |
message SearchResponse { | |
repeated Result results = 1; | |
} | |
message Result { | |
string url = 1; | |
string title = 2; | |
repeated string snippets = 3; | |
} | |
message Empty { | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#! /usr/bin/python | |
import os | |
import sys | |
import re | |
import json | |
def log(*msgs, sep=" ", end="\n"): | |
line = (sep.join(["{}".format(msg) for msg in msgs]))+end | |
sys.stderr.write(line) | |
sys.stderr.flush() | |
if len(sys.argv)!=2: | |
script=os.path.basename(sys.argv[0]) | |
log(f"""USAGE: {script} file.js > out.proto""") | |
exit(1) | |
print('''syntax = "proto3"; | |
import "google/protobuf/timestamp.proto"; | |
''') | |
sym_re = re.compile(r'exportSymbol\("(proto.[\w\.]+)"') | |
obj2_re = re.compile(r'(proto\.[\w\.]+?)(?:\.prototype)?\.toObject\s*=\s*function\((?:[^\)]*)\)\s*\{(?:[^\{\}]*?)(\{[^\}]*\})', re.S|re.M) | |
enum_re = re.compile(r'(proto\.[\w\.]+?)\s*=\s*(\{[^\{\}]*\})', re.S|re.M) | |
key_re = re.compile(r'\s*,?\s*([a-zA-Z]\w*)\s*:\s*', re.S|re.M) | |
enumerateMessagesSearchString = 'exportSymbol("proto.' | |
#obj_reader_re = re.compile(r'(proto\.[\w\.]+?)(?:\.prototype)?\.deserializeBinaryFromReader\s*=\s*function\((?:[^\)]*)\)\s*\{([^\}]*)\}', re.S|re.M) | |
obj_reader_re = re.compile(r'(proto\.[\w\.]+?)(?:\.prototype)?\.deserializeBinaryFromReader\s*=\s*function\((?:[^\)]*)\)\s*\{(.*?)\}[\s;]*return[^}]*}', re.S|re.M) | |
reader_parts_re = re.compile(r'case\s*(\d+)\s*:(.*?break;)', re.S|re.M) | |
reader_parts2_re = re.compile(r'.*?\.read(\w+)\(.*?\).*?(set|add)(\w+)\(', re.S|re.M) | |
reader_proto_re = re.compile(r'new\s*[\w\.]+\.([\w]+)', re.S|re.M) | |
reader_map_re = re.compile(r'\.read(Message\s*,\s*[\w\.]+\.(?:[\w]+)|[\w]+)', re.S|re.M) | |
reader_map_proto_re = re.compile(r'[\w\.]+\.([\w]+)\.deserialize', re.S|re.M) | |
obj_writer_re = re.compile(r'(proto\.[\w\.]+?)(?:\.prototype)?\.serializeBinaryToWriter\s*=\s*function\((?:[^\)]*)\)\s*\{([^\}]*)\}', re.S|re.M) | |
writer_parts_re = re.compile(r'\.write(\w+)\s*\(\s*(\d+)\s*,\s*[^)]*\)', re.S|re.M) | |
def lower1st(s): | |
if s is None: return None | |
if not s: return '' | |
return s[0].lower() + s[1:] | |
def upper1st(s): | |
if s is None: return None | |
if not s: return '' | |
return s[0].upper() + s[1:] | |
with open(sys.argv[1], 'r') as f: | |
jsInput = f.read() | |
syms = [ m for m in sym_re.findall(jsInput) ] | |
to_objs = { m: o for m, o in obj2_re.findall(jsInput) } | |
obj_readers = { m: reader_parts_re.findall(o) for m, o in obj_reader_re.findall(jsInput) } | |
obj_writers = { m: writer_parts_re.findall(o) for m, o in obj_writer_re.findall(jsInput) } | |
to_enums = { m: o for m, o in enum_re.findall(jsInput) } | |
to_enums_lower = { m.lower(): o for m, o in enum_re.findall(jsInput) } | |
done = set() | |
for m in syms: | |
last_dot = m.split('.')[-1] | |
if last_dot in done: | |
log(f"** WW **: {m} already defined as {last_dot}") | |
continue | |
done.add(last_dot) | |
enum_def = to_enums.get(m, None) | |
obj_def = to_objs.get(m, None) | |
if enum_def: | |
continue | |
enum_def = enum_def.replace(",", ";\n").replace("}", "};\n") | |
res[m] = f"enum {m} {enum_def}" | |
print(m, enum_def) | |
elif obj_def: | |
#print("# ", m) | |
ls = key_re.sub(r"\n\1: ", obj_def.strip("{}").replace("\n", " ")).split("\n") | |
ls = [ i.strip().split(':', 1) for i in ls if i.strip()] | |
reader_parts = {int(ix): desc for ix, desc in (obj_readers.get(m, None) or [])} | |
counter = max(reader_parts.keys() or [0]) + 1 | |
reserved = set(range(1, counter))-set(reader_parts.keys()) | |
#log("## II ## reserved: ", reserved) | |
#log("## II ## ls: ", ls) | |
#log("## II ## ls keys: ", [ a[0] for a in ls]) | |
#log("## II ## reader parts: ", reader_parts) | |
#log("## II ## writer parts: ", obj_writers.get(m, None)) | |
if (len(ls)>len(reader_parts)): | |
log("## EE ##", len(ls), len(reader_parts)) | |
exit(1) | |
it = iter(ls) | |
print(f'message {last_dot} {{') | |
for ix in range(1, counter): | |
if ix in reserved: continue | |
k, v = next(it) | |
k = k.strip() | |
v = v.strip() | |
desc = reader_parts[ix] | |
#print("** key=", k, ", v=", v, ", desc=",desc) | |
match = reader_parts2_re.search(desc) | |
if match: | |
typ, add_or_set, name = match.groups() | |
typ = lower1st(typ) | |
name = lower1st(name) | |
if name!=k and k.endswith('List'): | |
k = k[:-4] | |
if name!=k: | |
log("** EE ** mismatched name: ", name, k) | |
exit(1) | |
if typ=='message': | |
match2 = reader_proto_re.search(desc) | |
if match2: | |
typ = match2[1] | |
else: | |
log(" ** EE2 ** desc=", desc) | |
exit(1) | |
if add_or_set == 'add': typ='repeated '+typ | |
if typ=='enum': | |
typ = upper1st(k) | |
enum_k = m+'.'+typ | |
enum_def = to_enums.get(enum_k, None) or to_enums_lower.get(enum_k.lower(), None) | |
if not enum_def: | |
log("** WW ** enum not found:", m, k) | |
print(f" int32 {k} = {ix};") | |
continue | |
enum_def = enum_def.replace(",", ";\n").replace("}", ";\n}\n").replace("{", "{\n").replace(":", " = ") | |
print(f" enum {typ} {enum_def}") | |
print(f' {typ} {k} = {ix};') | |
elif "Map.deserializeBinary" in desc: | |
_, desc = desc.split('Map.deserializeBinary', 1) | |
map_typs = reader_map_re.findall(desc) | |
if not map_typs or len(map_typs)!=2: | |
print("** EE **") | |
map_from, map_to = map_typs | |
map_from = lower1st(map_from) | |
map_to = lower1st(map_to) | |
if map_from.startswith('message'): | |
match2 = reader_map_proto_re.search(map_from) | |
if match2: | |
map_from = match2[1] | |
if map_to.startswith('message'): | |
match2 = reader_map_proto_re.search(map_to) | |
if match2: | |
map_to = match2[1] | |
print(f' map<{map_from},{map_to}> {k} = {ix};') | |
else: | |
log("** EE **") | |
exit(1) | |
if reserved: print(' reserved '+', '.join([str(i) for i in sorted(reserved)])+';') | |
print(f'}}') | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment