Compare commits
28 Commits
v0.0.1
...
2011b828db
Author | SHA1 | Date | |
---|---|---|---|
2011b828db
|
|||
![]() |
a267bfb47f | ||
![]() |
8b0ea9450e | ||
d4f057d91c
|
|||
![]() |
7d2039d24e | ||
![]() |
e64ea4bdec | ||
![]() |
4badfa6321 | ||
![]() |
6e0089e01c | ||
![]() |
3afcfbc6a7 | ||
![]() |
aa292c7682 | ||
![]() |
5fe61462a2 | ||
![]() |
ad5f39db56 | ||
![]() |
97ec04789d | ||
![]() |
6dae4e801f | ||
![]() |
040732920b | ||
![]() |
a93fd14672 | ||
![]() |
a4a6a24f3e | ||
![]() |
307e0b8fd7 | ||
71f9fea419
|
|||
![]() |
7ff305a0f8 | ||
![]() |
665ec7efd6 | ||
![]() |
b404b47d74 | ||
![]() |
4363ff705c | ||
352d14ec82
|
|||
e7397945d3
|
|||
48cab893c9
|
|||
64df67ac8a
|
|||
d9e00c3762
|
55
.github/workflows/build.yml
vendored
Normal file
55
.github/workflows/build.yml
vendored
Normal file
@@ -0,0 +1,55 @@
|
||||
name: Build
|
||||
on:
|
||||
push:
|
||||
branches: [main]
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ${{ matrix.os }}
|
||||
strategy:
|
||||
matrix:
|
||||
os: [ubuntu-latest, macos-latest]
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Build
|
||||
run: |
|
||||
mkdir build
|
||||
cd build
|
||||
cmake ..
|
||||
cmake --build . --config Debug
|
||||
cmake --install .
|
||||
|
||||
- name: Upload artifact
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: pyarmor-1shot-build-${{ matrix.os }}
|
||||
path: |
|
||||
helpers
|
||||
README.md
|
||||
README-Decompyle++.markdown
|
||||
LICENSE
|
||||
|
||||
windows-build:
|
||||
runs-on: windows-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Build
|
||||
run: |
|
||||
mkdir build
|
||||
cd build
|
||||
cmake -G "MinGW Makefiles" ..
|
||||
cmake --build . --config Debug
|
||||
cmake --install .
|
||||
|
||||
- name: Upload artifact
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: pyarmor-1shot-build-windows
|
||||
path: |
|
||||
helpers
|
||||
README.md
|
||||
README-Decompyle++.markdown
|
||||
LICENSE
|
67
.github/workflows/codeql-analysis.yml
vendored
67
.github/workflows/codeql-analysis.yml
vendored
@@ -1,67 +0,0 @@
|
||||
# For most projects, this workflow file will not need changing; you simply need
|
||||
# to commit it to your repository.
|
||||
#
|
||||
# You may wish to alter this file to override the set of languages analyzed,
|
||||
# or to provide custom queries or build logic.
|
||||
name: "CodeQL"
|
||||
|
||||
on:
|
||||
push:
|
||||
branches: [master]
|
||||
pull_request:
|
||||
# The branches below must be a subset of the branches above
|
||||
branches: [master]
|
||||
schedule:
|
||||
- cron: '0 1 * * 2'
|
||||
|
||||
jobs:
|
||||
analyze:
|
||||
name: Analyze
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
actions: read
|
||||
contents: read
|
||||
security-events: write
|
||||
|
||||
strategy:
|
||||
fail-fast: false
|
||||
matrix:
|
||||
language: ['cpp', 'python']
|
||||
# CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
|
||||
# Use only 'java' to analyze code written in Java, Kotlin or both
|
||||
# Use only 'javascript' to analyze code written in JavaScript, TypeScript or both
|
||||
# Learn more about CodeQL language support at https://aka.ms/codeql-docs/language-support
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v3
|
||||
|
||||
# Initializes the CodeQL tools for scanning.
|
||||
- name: Initialize CodeQL
|
||||
uses: github/codeql-action/init@v2
|
||||
with:
|
||||
languages: ${{ matrix.language }}
|
||||
# If you wish to specify custom queries, you can do so here or in a config file.
|
||||
# By default, queries listed here will override any specified in a config file.
|
||||
# Prefix the list here with "+" to use these queries and those in the config file.
|
||||
|
||||
# Details on CodeQL's query packs refer to : https://docs.github.com/en/code-security/code-scanning/automatically-scanning-your-code-for-vulnerabilities-and-errors/configuring-code-scanning#using-queries-in-ql-packs
|
||||
# queries: security-extended,security-and-quality
|
||||
|
||||
# Autobuild attempts to build any compiled languages (C/C++, C#, Go, or Java).
|
||||
# If this step fails, then you should remove it and run the build manually (see below)
|
||||
- if: matrix.language == 'python'
|
||||
name: Autobuild Python
|
||||
uses: github/codeql-action/autobuild@v2
|
||||
|
||||
- if: matrix.language == 'cpp'
|
||||
name: Build C++
|
||||
run: |
|
||||
mkdir build && cd build
|
||||
cmake -DCMAKE_BUILD_TYPE=RelWithDebInfo ..
|
||||
make
|
||||
|
||||
- name: Perform CodeQL Analysis
|
||||
uses: github/codeql-action/analyze@v2
|
||||
with:
|
||||
category: "/language:${{matrix.language}}"
|
29
.github/workflows/linux-ci.yml
vendored
29
.github/workflows/linux-ci.yml
vendored
@@ -1,29 +0,0 @@
|
||||
name: Linux-CI
|
||||
on:
|
||||
push:
|
||||
branches: [master]
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v1
|
||||
- name: Configure and Build
|
||||
run: |
|
||||
(
|
||||
mkdir build-debug && cd build-debug
|
||||
cmake -DCMAKE_BUILD_TYPE=Debug ..
|
||||
make -j4
|
||||
)
|
||||
|
||||
(
|
||||
mkdir build-release && cd build-release
|
||||
cmake -DCMAKE_BUILD_TYPE=Debug ..
|
||||
make -j4
|
||||
)
|
||||
|
||||
- name: Test
|
||||
run: |
|
||||
cmake --build build-debug --target check
|
||||
cmake --build build-release --target check
|
29
.github/workflows/msvc-ci.yml
vendored
29
.github/workflows/msvc-ci.yml
vendored
@@ -1,29 +0,0 @@
|
||||
name: MSVC-CI
|
||||
on:
|
||||
push:
|
||||
branches: [master]
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: windows-latest
|
||||
steps:
|
||||
- uses: actions/checkout@v2
|
||||
- name: Configure and Build
|
||||
run: |
|
||||
mkdir build
|
||||
cd build
|
||||
cmake -G "Visual Studio 17 2022" -A Win32 ..
|
||||
cmake --build . --config Debug
|
||||
cmake --build . --config Release
|
||||
|
||||
- name: Test
|
||||
run: |
|
||||
cmake --build build --config Debug --target check
|
||||
cmake --build build --config Release --target check
|
||||
|
||||
- name: Upload artifact
|
||||
uses: actions/upload-artifact@v3
|
||||
with:
|
||||
name: pycdc-release
|
||||
path: build\Release\*.exe
|
4
.gitignore
vendored
4
.gitignore
vendored
@@ -6,8 +6,8 @@
|
||||
/.kdev4
|
||||
__pycache__
|
||||
tests-out
|
||||
|
||||
.vscode/
|
||||
build/
|
||||
.vscode/
|
||||
|
||||
pyarmor-1shot
|
||||
pyarmor-1shot.exe
|
||||
|
134
ASTree.cpp
134
ASTree.cpp
@@ -1113,7 +1113,10 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
case Pyc::INSTRUMENTED_FOR_ITER_A:
|
||||
{
|
||||
PycRef<ASTNode> iter = stack.top(); // Iterable
|
||||
stack.pop();
|
||||
if (mod->verCompare(3, 12) < 0) {
|
||||
// Do not pop the iterator for py 3.12+
|
||||
stack.pop();
|
||||
}
|
||||
/* Pop it? Don't pop it? */
|
||||
|
||||
int end;
|
||||
@@ -1379,10 +1382,13 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
}
|
||||
break;
|
||||
case Pyc::JUMP_ABSOLUTE_A:
|
||||
// bpo-47120: Replaced JUMP_ABSOLUTE by the relative jump JUMP_BACKWARD.
|
||||
case Pyc::JUMP_BACKWARD_A:
|
||||
case Pyc::JUMP_BACKWARD_NO_INTERRUPT_A:
|
||||
{
|
||||
int offs = operand;
|
||||
if (mod->verCompare(3, 10) >= 0)
|
||||
offs *= sizeof(uint16_t); // // BPO-27129
|
||||
offs *= sizeof(uint16_t); // // BPO-27129
|
||||
|
||||
if (offs < pos) {
|
||||
if (curblock->blktype() == ASTBlock::BLK_FOR) {
|
||||
@@ -1873,6 +1879,13 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
mod->verCompare(3, 11) >= 0 ? code->qualName()->value() : code->name()->value());
|
||||
}
|
||||
}
|
||||
// BEGIN ONESHOT TEMPORARY PATCH
|
||||
// The problem is not here, but in the way the blocks are created
|
||||
// Add this to avoid segfault
|
||||
if (blocks.top() == defblock)
|
||||
break;
|
||||
// END ONESHOT PATCH
|
||||
|
||||
PycRef<ASTBlock> tmp = curblock;
|
||||
blocks.pop();
|
||||
|
||||
@@ -1942,10 +1955,37 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
case Pyc::POP_EXCEPT:
|
||||
/* Do nothing. */
|
||||
break;
|
||||
case Pyc::END_FOR:
|
||||
{
|
||||
stack.pop();
|
||||
|
||||
if ((opcode == Pyc::END_FOR) && (mod->majorVer() == 3) && (mod->minorVer() == 12)) {
|
||||
// one additional pop for python 3.12
|
||||
stack.pop();
|
||||
}
|
||||
|
||||
// end for loop here
|
||||
/* TODO : Ensure that FOR loop ends here.
|
||||
Due to CACHE instructions at play, the end indicated in
|
||||
the for loop by pycdas is not correct, it is off by
|
||||
some small amount. */
|
||||
if (curblock->blktype() == ASTBlock::BLK_FOR) {
|
||||
PycRef<ASTBlock> prev = blocks.top();
|
||||
blocks.pop();
|
||||
|
||||
curblock = blocks.top();
|
||||
curblock->append(prev.cast<ASTNode>());
|
||||
}
|
||||
else {
|
||||
fprintf(stderr, "Wrong block type %i for END_FOR\n", curblock->blktype());
|
||||
}
|
||||
}
|
||||
break;
|
||||
case Pyc::POP_TOP:
|
||||
{
|
||||
PycRef<ASTNode> value = stack.top();
|
||||
stack.pop();
|
||||
|
||||
if (!curblock->inited()) {
|
||||
if (curblock->blktype() == ASTBlock::BLK_WITH) {
|
||||
curblock.cast<ASTWithBlock>()->setExpr(value);
|
||||
@@ -2051,8 +2091,6 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
blocks.pop();
|
||||
curblock = blocks.top();
|
||||
curblock->append(prev.cast<ASTNode>());
|
||||
|
||||
bc_next(source, mod, opcode, operand, pos);
|
||||
}
|
||||
}
|
||||
break;
|
||||
@@ -2749,21 +2787,6 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
break;
|
||||
// BEGIN ONESHOT TEMPORARY PATCH
|
||||
// THESE OPCODES ARE NOT IMPLEMENTED HERE
|
||||
case Pyc::COPY_A:
|
||||
{
|
||||
FastStack tmp_stack(20);
|
||||
for (int i = 0; i < operand - 1; i++) {
|
||||
tmp_stack.push(stack.top());
|
||||
stack.pop();
|
||||
}
|
||||
auto value = stack.top();
|
||||
for (int i = 0; i < operand - 1; i++) {
|
||||
stack.push(tmp_stack.top());
|
||||
tmp_stack.pop();
|
||||
}
|
||||
stack.push(value);
|
||||
}
|
||||
break;
|
||||
case Pyc::PUSH_EXC_INFO:
|
||||
{
|
||||
stack.push(stack.top());
|
||||
@@ -2790,6 +2813,79 @@ PycRef<ASTNode> BuildFromCode(PycRef<PycCode> code, PycModule* mod)
|
||||
}
|
||||
break;
|
||||
// END ONESHOT PATCH
|
||||
case Pyc::BINARY_SLICE:
|
||||
{
|
||||
PycRef<ASTNode> end = stack.top();
|
||||
stack.pop();
|
||||
PycRef<ASTNode> start = stack.top();
|
||||
stack.pop();
|
||||
PycRef<ASTNode> dest = stack.top();
|
||||
stack.pop();
|
||||
|
||||
if (start.type() == ASTNode::NODE_OBJECT
|
||||
&& start.cast<ASTObject>()->object() == Pyc_None) {
|
||||
start = NULL;
|
||||
}
|
||||
|
||||
if (end.type() == ASTNode::NODE_OBJECT
|
||||
&& end.cast<ASTObject>()->object() == Pyc_None) {
|
||||
end = NULL;
|
||||
}
|
||||
|
||||
PycRef<ASTNode> slice;
|
||||
if (start == NULL && end == NULL) {
|
||||
slice = new ASTSlice(ASTSlice::SLICE0);
|
||||
} else if (start == NULL) {
|
||||
slice = new ASTSlice(ASTSlice::SLICE2, start, end);
|
||||
} else if (end == NULL) {
|
||||
slice = new ASTSlice(ASTSlice::SLICE1, start, end);
|
||||
} else {
|
||||
slice = new ASTSlice(ASTSlice::SLICE3, start, end);
|
||||
}
|
||||
stack.push(new ASTSubscr(dest, slice));
|
||||
}
|
||||
break;
|
||||
case Pyc::STORE_SLICE:
|
||||
{
|
||||
PycRef<ASTNode> end = stack.top();
|
||||
stack.pop();
|
||||
PycRef<ASTNode> start = stack.top();
|
||||
stack.pop();
|
||||
PycRef<ASTNode> dest = stack.top();
|
||||
stack.pop();
|
||||
PycRef<ASTNode> values = stack.top();
|
||||
stack.pop();
|
||||
|
||||
if (start.type() == ASTNode::NODE_OBJECT
|
||||
&& start.cast<ASTObject>()->object() == Pyc_None) {
|
||||
start = NULL;
|
||||
}
|
||||
|
||||
if (end.type() == ASTNode::NODE_OBJECT
|
||||
&& end.cast<ASTObject>()->object() == Pyc_None) {
|
||||
end = NULL;
|
||||
}
|
||||
|
||||
PycRef<ASTNode> slice;
|
||||
if (start == NULL && end == NULL) {
|
||||
slice = new ASTSlice(ASTSlice::SLICE0);
|
||||
} else if (start == NULL) {
|
||||
slice = new ASTSlice(ASTSlice::SLICE2, start, end);
|
||||
} else if (end == NULL) {
|
||||
slice = new ASTSlice(ASTSlice::SLICE1, start, end);
|
||||
} else {
|
||||
slice = new ASTSlice(ASTSlice::SLICE3, start, end);
|
||||
}
|
||||
|
||||
curblock->append(new ASTStore(values, new ASTSubscr(dest, slice)));
|
||||
}
|
||||
break;
|
||||
case Pyc::COPY_A:
|
||||
{
|
||||
PycRef<ASTNode> value = stack.top(operand);
|
||||
stack.push(value);
|
||||
}
|
||||
break;
|
||||
default:
|
||||
fprintf(stderr, "Unsupported opcode: %s (%d) at %s\n",
|
||||
Pyc::OpcodeName(opcode),
|
||||
|
24
FastStack.h
24
FastStack.h
@@ -30,14 +30,30 @@ public:
|
||||
{
|
||||
if (m_ptr > -1)
|
||||
m_stack[m_ptr--] = nullptr;
|
||||
else {
|
||||
#ifdef BLOCK_DEBUG
|
||||
fprintf(stderr, "pop from empty stack\n");
|
||||
#endif
|
||||
}
|
||||
}
|
||||
|
||||
PycRef<ASTNode> top() const
|
||||
PycRef<ASTNode> top(int i = 1) const
|
||||
{
|
||||
if (m_ptr > -1)
|
||||
return m_stack[m_ptr];
|
||||
else
|
||||
if (i > 0) {
|
||||
int idx = m_ptr + 1 - i;
|
||||
if ((m_ptr > -1) && (idx >= 0))
|
||||
return m_stack[idx];
|
||||
else {
|
||||
#ifdef BLOCK_DEBUG
|
||||
fprintf(stderr, "insufficient values on stack\n");
|
||||
#endif
|
||||
return nullptr;
|
||||
}
|
||||
}
|
||||
else {
|
||||
fprintf(stderr, "incorrect operand %i\n", i);
|
||||
return nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
bool empty() const
|
||||
|
50
README.md
50
README.md
@@ -1,12 +1,32 @@
|
||||
# Pyarmor-Static-Unpack-1shot
|
||||
# Pyarmor Static Unpack One-Shot Tool
|
||||
|
||||
Generally this project aims to statically convert (without executing) armored data - which can be regarded as an encrypted variant of pyc files - back to disassembly and (experimentally) source code. Therefore we forked the awesome [Decompyle++](https://github.com/zrax/pycdc) (aka pycdc).
|
||||
[Pyarmor](https://github.com/dashingsoft/pyarmor) is a popular tool to protect Python source code. It turns Python scripts into binary data, which can be regarded as an encrypted variant of pyc files. They can be decrypted by a shared library (pyarmor_runtime) and then executed by Python interpreter.
|
||||
|
||||
Currently we are trying to support Pyarmor 8.0 - latest (9.1.1), Python 3.7 - 3.13, platforms covering Windows, Linux, macOS, and Android, with obfuscating options as many as possible. (However, we only have limited tests.)
|
||||
This project aims to convert armored data back to bytecode assembly and (experimentally) source code. We forked the awesome [Decompyle++](https://github.com/zrax/pycdc) (aka pycdc), and added some processes on it like modifying abstract syntax tree.
|
||||
|
||||
If the data starts with `PY` followed by six digits, it is supported. Otherwise, if it starts with `PYARMOR`, it is generated by Pyarmor 7 or before, and is not supported.
|
||||
> [!WARNING]
|
||||
>
|
||||
> **Disassembly results are accurate, but decompiled code can be incomplete and incorrect.** [See issue #3](https://github.com/Lil-House/Pyarmor-Static-Unpack-1shot/issues/3)
|
||||
|
||||
We cannot wait to make it public. Detailed write-up will be available soon. For those who are curious, temporarily you can check out [the similar work of G DATA Advanced Analytics](https://cyber.wtf/2025/02/12/unpacking-pyarmor-v8-scripts/).
|
||||
## Features
|
||||
|
||||
### Static
|
||||
|
||||
You don't need to execute the encrypted script. We decrypt them using the same algorithm as pyarmor_runtime. This is useful when the scripts cannot be trusted.
|
||||
|
||||
### Universal
|
||||
|
||||
Currently we are trying to support Pyarmor 8.0 to 9.1.3 (latest), Python 3.7 - 3.13, on all operating systems, with obfuscating options as many as possible. (However, we only have limited tests.)
|
||||
|
||||
You can run this tool in any environment, no need to be the same with obfuscated scripts or runtime.
|
||||
|
||||
> [!NOTE]
|
||||
>
|
||||
> If the data starts with `PY` followed by six digits, it is supported. Otherwise, if it starts with `PYARMOR`, it is generated by Pyarmor 7 or earlier, and is not supported.
|
||||
|
||||
### Easy to use
|
||||
|
||||
The only thing you need to do is specifying where your obfuscated scripts are. The tool does everything like detecting armored data, parsing, disassembling, and decompiling. See "Usage" section below.
|
||||
|
||||
## Build
|
||||
|
||||
@@ -18,23 +38,25 @@ cmake --build .
|
||||
cmake --install .
|
||||
```
|
||||
|
||||
You can also download prebuilt binary files on [releases page](https://github.com/Lil-House/Pyarmor-Static-Unpack-1shot/releases).
|
||||
|
||||
## Usage
|
||||
|
||||
Make sure the executable `pyarmor-1shot` (`pyarmor-1shot.exe` on Windows) exists in `helpers` directory, and run `helpers/shot.py` in Python 3 (no need to use the same version with obfuscated scripts) with the "root" directory of obfuscated scripts. It will recursively find and handle `pyarmor_runtime` and as much armored data as possible. For example:
|
||||
|
||||
``` bash
|
||||
$ ls /path/to/scripts
|
||||
__pycache__ pyarmor_runtime_000000 obf_main.py plain_src.py util.pyc packed.so folder_with_other_scripts readme.unrelated
|
||||
$ python /path/to/helpers/shot.py /path/to/scripts
|
||||
python /path/to/helpers/shot.py /path/to/scripts
|
||||
```
|
||||
|
||||
Before running `shot.py`, make sure the executable `pyarmor-1shot` (`pyarmor-1shot.exe` on Windows) exists in `helpers` directory.
|
||||
|
||||
You only need to specify the directory that contains all armored data and `pyarmor_runtime`. The tool finds and handles them recursively as much as possible.
|
||||
|
||||
When necessary, specify a `pyarmor_runtime` executable with `-r path/to/pyarmor_runtime[.pyd|.so|.dylib]`.
|
||||
|
||||
All files generated from this tool have a `.1shot.` in file names. If you want to save them in another directory instead of in-place, use `-o another/path/`. Folder structure will remain unchanged.
|
||||
|
||||
Note:
|
||||
|
||||
- Subdirectories called `__pycache__` or `site-packages` will not be touched, and symbolic links will not be followed, to avoid repeat or forever loop and save time. If you really need them, run the script later in these directories (as "root" directory) and specify the runtime.
|
||||
- Subdirectories will not be touched if the folder name is exactly `__pycache__` or `site-packages` or it directly contains a file named `.no1shot`, and symbolic links will not be followed, to avoid repeat or forever loop and save time. If you really need them, run the script later in these directories and specify the runtime.
|
||||
- Archives, executables generated by PyInstaller and so on, must be unpacked by other tools before decrypting, or you will encounter undefined behavior.
|
||||
|
||||
## Feedback
|
||||
@@ -43,10 +65,6 @@ Feel free to open an issue if you have any questions, suggestions, or problems.
|
||||
|
||||
## Todo (PR Welcome!)
|
||||
|
||||
- [ ] Write-up
|
||||
- [ ] Multi-platform pyarmor_runtime executable
|
||||
- [ ] Accept more input forms
|
||||
- [ ] Tests for different Pyarmor and Python versions
|
||||
- [ ] Support more obfuscating options
|
||||
- [ ] Use asyncio for concurrency
|
||||
- [ ] Pyarmor 7 and before (Later or never.)
|
||||
- [ ] Regenerate pyc for other backends
|
||||
|
@@ -474,6 +474,10 @@ void bc_disasm(std::ostream& pyc_output, PycRef<PycCode> code, PycModule* mod,
|
||||
case Pyc::INSTRUMENTED_POP_JUMP_IF_FALSE_A:
|
||||
case Pyc::INSTRUMENTED_POP_JUMP_IF_TRUE_A:
|
||||
{
|
||||
/* TODO: Fix offset based on CACHE instructions.
|
||||
Offset is relative to next non-CACHE instruction
|
||||
and thus will be printed lower than actual value.
|
||||
See TODO @ END_FOR ASTree.cpp */
|
||||
int offs = operand;
|
||||
if (mod->verCompare(3, 10) >= 0)
|
||||
offs *= sizeof(uint16_t); // BPO-27129
|
||||
|
157
helpers/detect.py
Normal file
157
helpers/detect.py
Normal file
@@ -0,0 +1,157 @@
|
||||
import logging
|
||||
import os
|
||||
from typing import List, Tuple, Union
|
||||
|
||||
|
||||
def ascii_ratio(data: bytes) -> float:
|
||||
return sum(32 <= c < 127 for c in data) / len(data)
|
||||
|
||||
|
||||
def source_as_file(file_path: str) -> Union[List[bytes], None]:
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
co = compile(f.read(), '<str>', 'exec')
|
||||
data = [i for i in co.co_consts if type(i) is bytes
|
||||
and i.startswith(b'PY00') and len(i) > 64]
|
||||
return data
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
def source_as_lines(file_path: str) -> Union[List[bytes], None]:
|
||||
data = []
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
for line in f:
|
||||
try:
|
||||
co = compile(line, '<str>', 'exec')
|
||||
data.extend([i for i in co.co_consts if type(i) is bytes
|
||||
and i.startswith(b'PY00') and len(i) > 64])
|
||||
except:
|
||||
# ignore not compilable lines
|
||||
pass
|
||||
except:
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def find_data_from_bytes(data: bytes, max_count=-1) -> List[bytes]:
|
||||
result = []
|
||||
idx = 0
|
||||
while len(result) != max_count:
|
||||
idx = data.find(b'PY00')
|
||||
if idx == -1:
|
||||
break
|
||||
data = data[idx:]
|
||||
if len(data) < 64:
|
||||
break
|
||||
header_len = int.from_bytes(data[28:32], 'little')
|
||||
body_len = int.from_bytes(data[32:36], 'little')
|
||||
if header_len > 256 or body_len > 0xFFFFF or header_len + body_len > len(data):
|
||||
# compressed or coincident, skip
|
||||
data = data[5:]
|
||||
continue
|
||||
result.append(data[:header_len + body_len])
|
||||
|
||||
# maybe followed by data for other Python versions from the same file,
|
||||
# we do not extract them
|
||||
followed_by_another_equivalent = int.from_bytes(
|
||||
data[56:60], 'little') != 0
|
||||
data = data[header_len + body_len:]
|
||||
while followed_by_another_equivalent \
|
||||
and data.startswith(b'PY00') \
|
||||
and len(data) >= 64:
|
||||
header_len = int.from_bytes(data[28:32], 'little')
|
||||
body_len = int.from_bytes(data[32:36], 'little')
|
||||
followed_by_another_equivalent = int.from_bytes(
|
||||
data[56:60], 'little') != 0
|
||||
data = data[header_len + body_len:]
|
||||
return result
|
||||
|
||||
|
||||
def nuitka_package(head: bytes, relative_path: str) -> Union[List[Tuple[str, bytes]], None]:
|
||||
first_occurrence = head.find(b'PY00')
|
||||
if first_occurrence == -1:
|
||||
return None
|
||||
last_dot_bytecode = head.rfind(b'.bytecode\x00', 0, first_occurrence)
|
||||
if last_dot_bytecode == -1:
|
||||
return None
|
||||
length = int.from_bytes(
|
||||
head[last_dot_bytecode-4:last_dot_bytecode], 'little')
|
||||
end = last_dot_bytecode + length
|
||||
cur = last_dot_bytecode
|
||||
result = []
|
||||
while cur < end:
|
||||
module_name_len = head.find(b'\x00', cur, end) - cur
|
||||
module_name = head[cur:cur + module_name_len].decode('utf-8')
|
||||
cur += module_name_len + 1
|
||||
module_len = int.from_bytes(head[cur:cur + 4], 'little')
|
||||
cur += 4
|
||||
module_data = find_data_from_bytes(head[cur:cur + module_len], 1)
|
||||
if module_data:
|
||||
result.append((os.path.join(relative_path.rstrip(
|
||||
'/\\') + '.1shot.ext', module_name), module_data[0]))
|
||||
cur += module_len
|
||||
if result:
|
||||
logger = logging.getLogger('detect')
|
||||
logger.info(f'Found data in Nuitka package: {relative_path}')
|
||||
return result
|
||||
return None
|
||||
|
||||
|
||||
def detect_process(file_path: str, relative_path: str) -> Union[List[Tuple[str, bytes]], None]:
|
||||
'''
|
||||
Returns a list of (relative_path, bytes_raw) tuples, or None.
|
||||
Do not raise exceptions.
|
||||
'''
|
||||
logger = logging.getLogger('detect')
|
||||
|
||||
try:
|
||||
with open(file_path, 'rb') as f:
|
||||
head = f.read(16 * 1024 * 1024)
|
||||
except:
|
||||
logger.error(f'Failed to read file: {relative_path}')
|
||||
return None
|
||||
|
||||
if b'__pyarmor__' not in head:
|
||||
# no need to dig deeper
|
||||
return None
|
||||
|
||||
if ascii_ratio(head[:2048]) >= 0.9:
|
||||
# the whole file may not be compiled, but we can still try some lines;
|
||||
# None means failure (then we make another try),
|
||||
# empty list means success but no data found (then we skip this file)
|
||||
result = source_as_file(file_path)
|
||||
if result is None:
|
||||
result = source_as_lines(file_path)
|
||||
if result is None:
|
||||
return None
|
||||
|
||||
result_len = len(result)
|
||||
if result_len == 0:
|
||||
return None
|
||||
elif result_len == 1:
|
||||
logger.info(f'Found data in source: {relative_path}')
|
||||
return [(relative_path, result[0])]
|
||||
else:
|
||||
logger.info(f'Found data in source: {relative_path}')
|
||||
return [(f'{relative_path}__{i}', result[i]) for i in range(len(result))]
|
||||
|
||||
# binary file
|
||||
# ignore data after 16MB, before we have a reason to read more
|
||||
|
||||
if b'Error, corrupted constants object' in head:
|
||||
# an interesting special case: packer put armored data in a Nuitka package
|
||||
# we can know the exact module names, instead of adding boring __0, __1, ...
|
||||
return nuitka_package(head, relative_path)
|
||||
|
||||
result = find_data_from_bytes(head)
|
||||
result_len = len(result)
|
||||
if result_len == 0:
|
||||
return None
|
||||
elif result_len == 1:
|
||||
logger.info(f'Found data in binary: {relative_path}')
|
||||
return [(relative_path, result[0])]
|
||||
else:
|
||||
logger.info(f'Found data in binary: {relative_path}')
|
||||
return [(f'{relative_path}__{i}', result[i]) for i in range(len(result))]
|
448
helpers/shot.py
448
helpers/shot.py
@@ -2,13 +2,24 @@ import argparse
|
||||
from Crypto.Cipher import AES
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import asyncio
|
||||
import traceback
|
||||
import platform
|
||||
from typing import Dict, List, Tuple
|
||||
|
||||
try:
|
||||
from colorama import init, Fore, Style
|
||||
except ImportError:
|
||||
def init(**kwargs): pass
|
||||
class Fore: CYAN = RED = YELLOW = GREEN = ''
|
||||
class Style: RESET_ALL = ''
|
||||
|
||||
from detect import detect_process
|
||||
from runtime import RuntimeInfo
|
||||
|
||||
|
||||
SUBPROCESS_TIMEOUT = 30
|
||||
# Initialize colorama
|
||||
init(autoreset=True)
|
||||
|
||||
|
||||
def general_aes_ctr_decrypt(data: bytes, key: bytes, nonce: bytes) -> bytes:
|
||||
@@ -16,75 +27,243 @@ def general_aes_ctr_decrypt(data: bytes, key: bytes, nonce: bytes) -> bytes:
|
||||
return cipher.decrypt(data)
|
||||
|
||||
|
||||
def decrypt_process(runtimes: Dict[str, RuntimeInfo], sequences: List[Tuple[str, bytes]], args):
|
||||
def bcc_fallback_method(seq_file_path: str, output_path: str = None):
|
||||
# Fallback method for BCC mode deobfuscation
|
||||
logger = logging.getLogger('shot')
|
||||
logger.info(f'{Fore.YELLOW}Attempting BCC fallback method for: {seq_file_path}{Style.RESET_ALL}')
|
||||
|
||||
if output_path is None:
|
||||
output_path = seq_file_path + '.back.1shot.seq'
|
||||
|
||||
try:
|
||||
with open(seq_file_path, 'rb') as f:
|
||||
origin = f.read()
|
||||
|
||||
one_shot_header = origin[:32] # Header format
|
||||
aes_key = one_shot_header[1:17]
|
||||
|
||||
bcc_part_length = int.from_bytes(origin[0x58:0x5C], 'little') # If it is 0, it is not BCC part but bytecode part
|
||||
bytecode_part = origin[32+bcc_part_length:]
|
||||
aes_nonce = bytecode_part[36:40] + bytecode_part[44:52] # The same position as non-BCC file
|
||||
|
||||
with open(output_path, 'wb') as f:
|
||||
f.write(one_shot_header)
|
||||
f.write(bytecode_part[:64])
|
||||
f.write(general_aes_ctr_decrypt(bytecode_part[64:], aes_key, aes_nonce))
|
||||
|
||||
logger.info(f'{Fore.GREEN}Successfully created BCC fallback file: {output_path}{Style.RESET_ALL}')
|
||||
return output_path
|
||||
except Exception as e:
|
||||
error_details = traceback.format_exc()
|
||||
logger.error(f'{Fore.RED}BCC fallback method failed: {e}{Style.RESET_ALL}')
|
||||
logger.error(f'{Fore.RED}Error details: {error_details}{Style.RESET_ALL}')
|
||||
return None
|
||||
|
||||
|
||||
async def run_subprocess_async(cmd, cwd=None):
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*cmd,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
cwd=cwd
|
||||
)
|
||||
stdout, stderr = await process.communicate()
|
||||
return process.returncode, stdout, stderr
|
||||
|
||||
|
||||
async def decrypt_file_async(exe_path, seq_file_path, path, args):
|
||||
logger = logging.getLogger('shot')
|
||||
try:
|
||||
# Run without timeout
|
||||
returncode, stdout, stderr = await run_subprocess_async([exe_path, seq_file_path])
|
||||
|
||||
stdout_lines = stdout.decode('latin-1').splitlines()
|
||||
stderr_lines = stderr.decode('latin-1').splitlines()
|
||||
|
||||
for line in stdout_lines:
|
||||
logger.warning(f'PYCDC: {line} ({path})')
|
||||
|
||||
for line in stderr_lines:
|
||||
if line.startswith((
|
||||
'Warning: Stack history is empty',
|
||||
'Warning: Stack history is not empty',
|
||||
'Warning: block stack is not empty',
|
||||
)):
|
||||
if args.show_warn_stack or args.show_all:
|
||||
logger.warning(f'PYCDC: {line} ({path})')
|
||||
elif line.startswith('Unsupported opcode:'):
|
||||
if args.show_err_opcode or args.show_all:
|
||||
logger.error(f'PYCDC: {line} ({path})')
|
||||
elif line.startswith((
|
||||
'Something TERRIBLE happened',
|
||||
'Unsupported argument',
|
||||
'Unsupported Node type',
|
||||
'Unsupported node type',
|
||||
)): # annoying wont-fix errors
|
||||
if args.show_all:
|
||||
logger.error(f'PYCDC: {line} ({path})')
|
||||
else:
|
||||
logger.error(f'PYCDC: {line} ({path})')
|
||||
|
||||
return returncode, stdout_lines, stderr_lines
|
||||
except Exception as e:
|
||||
error_details = traceback.format_exc()
|
||||
logger.error(f'{Fore.RED}Error during async deobfuscation: {e}{Style.RESET_ALL}')
|
||||
logger.error(f'{Fore.RED}Error details: {error_details}{Style.RESET_ALL}')
|
||||
return -1, [], []
|
||||
|
||||
|
||||
async def decrypt_process_async(runtimes: Dict[str, RuntimeInfo], sequences: List[Tuple[str, bytes]], args):
|
||||
logger = logging.getLogger('shot')
|
||||
output_dir: str = args.output_dir or args.directory
|
||||
for path, data in sequences:
|
||||
try:
|
||||
serial_number = data[2:8].decode('utf-8')
|
||||
runtime = runtimes[serial_number]
|
||||
logger.info(f'Decrypting: {serial_number} ({path})')
|
||||
|
||||
# Create a semaphore to limit concurrent processes
|
||||
semaphore = asyncio.Semaphore(args.concurrent) # Use the concurrent argument
|
||||
|
||||
# Get the appropriate executable for the current platform
|
||||
exe_path = get_platform_executable(args)
|
||||
|
||||
dest_path = os.path.join(output_dir, path) if output_dir else path
|
||||
dest_dir = os.path.dirname(dest_path)
|
||||
if not os.path.exists(dest_dir):
|
||||
os.makedirs(dest_dir)
|
||||
async def process_file(path, data):
|
||||
async with semaphore:
|
||||
try:
|
||||
serial_number = data[2:8].decode('utf-8')
|
||||
runtime = runtimes[serial_number]
|
||||
logger.info(f'{Fore.CYAN}Decrypting: {serial_number} ({path}){Style.RESET_ALL}')
|
||||
|
||||
if args.export_raw_data:
|
||||
with open(dest_path + '.1shot.raw', 'wb') as f:
|
||||
f.write(data)
|
||||
dest_path = os.path.join(output_dir, path) if output_dir else path
|
||||
dest_dir = os.path.dirname(dest_path)
|
||||
if not os.path.exists(dest_dir):
|
||||
os.makedirs(dest_dir)
|
||||
|
||||
cipher_text_offset = int.from_bytes(data[28:32], 'little')
|
||||
cipher_text_length = int.from_bytes(data[32:36], 'little')
|
||||
nonce = data[36:40] + data[44:52]
|
||||
with open(dest_path + '.1shot.seq', 'wb') as f:
|
||||
f.write(b'\xa1' + runtime.runtime_aes_key)
|
||||
f.write(b'\xa2' + runtime.mix_str_aes_nonce())
|
||||
f.write(b'\xf0\xff')
|
||||
f.write(data[:cipher_text_offset])
|
||||
f.write(general_aes_ctr_decrypt(
|
||||
data[cipher_text_offset:cipher_text_offset+cipher_text_length], runtime.runtime_aes_key, nonce))
|
||||
f.write(data[cipher_text_offset+cipher_text_length:])
|
||||
if args.export_raw_data:
|
||||
with open(dest_path + '.1shot.raw', 'wb') as f:
|
||||
f.write(data)
|
||||
|
||||
exe_name = 'pyarmor-1shot.exe' if os.name == 'nt' else 'pyarmor-1shot'
|
||||
exe_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)), exe_name)
|
||||
# TODO: multi process
|
||||
sp = subprocess.run(
|
||||
[
|
||||
exe_path,
|
||||
dest_path + '.1shot.seq',
|
||||
],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=SUBPROCESS_TIMEOUT,
|
||||
)
|
||||
stdout = sp.stdout.decode('latin-1').splitlines()
|
||||
stderr = sp.stderr.decode('latin-1').splitlines()
|
||||
for line in stdout:
|
||||
logger.warning(f'PYCDC: {line} ({path})')
|
||||
for line in stderr:
|
||||
if line.startswith((
|
||||
'Warning: Stack history is empty',
|
||||
'Warning: Stack history is not empty',
|
||||
'Warning: block stack is not empty',
|
||||
)):
|
||||
if args.show_warn_stack or args.show_all:
|
||||
logger.warning(f'PYCDC: {line} ({path})')
|
||||
elif line.startswith('Unsupported opcode:'):
|
||||
if args.show_err_opcode or args.show_all:
|
||||
logger.error(f'PYCDC: {line} ({path})')
|
||||
elif line.startswith('Something TERRIBLE happened'):
|
||||
if args.show_all:
|
||||
logger.error(f'PYCDC: {line} ({path})')
|
||||
cipher_text_offset = int.from_bytes(data[28:32], 'little')
|
||||
cipher_text_length = int.from_bytes(data[32:36], 'little')
|
||||
nonce = data[36:40] + data[44:52]
|
||||
seq_file_path = dest_path + '.1shot.seq'
|
||||
with open(seq_file_path, 'wb') as f:
|
||||
f.write(b'\xa1' + runtime.runtime_aes_key)
|
||||
f.write(b'\xa2' + runtime.mix_str_aes_nonce())
|
||||
f.write(b'\xf0\xff')
|
||||
f.write(data[:cipher_text_offset])
|
||||
f.write(general_aes_ctr_decrypt(
|
||||
data[cipher_text_offset:cipher_text_offset+cipher_text_length], runtime.runtime_aes_key, nonce))
|
||||
f.write(data[cipher_text_offset+cipher_text_length:])
|
||||
|
||||
# Run without timeout
|
||||
returncode, stdout_lines, stderr_lines = await decrypt_file_async(exe_path, seq_file_path, path, args)
|
||||
|
||||
# Check for specific errors that indicate BCC mode
|
||||
should_try_bcc = False
|
||||
error_message = ""
|
||||
|
||||
# FIXME: Probably false positive
|
||||
if returncode != 0:
|
||||
error_message = f"PYCDC returned {returncode}"
|
||||
# Check for specific error patterns that suggest BCC mode
|
||||
for line in stderr_lines:
|
||||
if ("Unsupported opcode" in line or
|
||||
"Something TERRIBLE happened" in line or
|
||||
"Unknown opcode 0" in line or
|
||||
"Got unsupported type" in line):
|
||||
should_try_bcc = True
|
||||
error_message += f" - {line}"
|
||||
break
|
||||
|
||||
if should_try_bcc:
|
||||
logger.warning(f'{Fore.YELLOW}{error_message} ({path}) - Attempting BCC fallback{Style.RESET_ALL}')
|
||||
# Try BCC fallback method
|
||||
bcc_file_path = bcc_fallback_method(seq_file_path)
|
||||
if bcc_file_path:
|
||||
logger.info(f'{Fore.GREEN}Running deobfuscator on BCC fallback file{Style.RESET_ALL}')
|
||||
try:
|
||||
# Run without timeout
|
||||
returncode, stdout_lines, stderr_lines = await decrypt_file_async(exe_path, bcc_file_path, path, args)
|
||||
if returncode == 0:
|
||||
logger.info(f'{Fore.GREEN}Successfully deobfuscated using BCC fallback method{Style.RESET_ALL}')
|
||||
print(f"{Fore.GREEN} BCC Decrypted: {path}{Style.RESET_ALL}")
|
||||
else:
|
||||
logger.error(f'{Fore.RED}BCC fallback deobfuscation failed with return code {returncode}{Style.RESET_ALL}')
|
||||
for line in stderr_lines:
|
||||
logger.error(f'{Fore.RED}BCC Error: {line}{Style.RESET_ALL}')
|
||||
except Exception as e:
|
||||
error_details = traceback.format_exc()
|
||||
logger.error(f'{Fore.RED}BCC fallback deobfuscation failed with error: {e}{Style.RESET_ALL}')
|
||||
logger.error(f'{Fore.RED}Error details: {error_details}{Style.RESET_ALL}')
|
||||
elif returncode == 0:
|
||||
# Successfully decrypted
|
||||
logger.info(f'{Fore.GREEN}Successfully decrypted: {path}{Style.RESET_ALL}')
|
||||
print(f"{Fore.GREEN} Decrypted: {path}{Style.RESET_ALL}")
|
||||
else:
|
||||
logger.error(f'PYCDC: {line} ({path})')
|
||||
if sp.returncode != 0:
|
||||
logger.warning(f'PYCDC returned {sp.returncode} ({path})')
|
||||
continue
|
||||
except Exception as e:
|
||||
logger.error(f'Decrypt failed: {e} ({path})')
|
||||
continue
|
||||
logger.warning(f'{Fore.YELLOW}{error_message} ({path}){Style.RESET_ALL}')
|
||||
except Exception as e:
|
||||
error_details = traceback.format_exc()
|
||||
logger.error(f'{Fore.RED}Decrypt failed: {e} ({path}){Style.RESET_ALL}')
|
||||
logger.error(f'{Fore.RED}Error details: {error_details}{Style.RESET_ALL}')
|
||||
|
||||
# Create tasks for all files
|
||||
tasks = [process_file(path, data) for path, data in sequences]
|
||||
|
||||
# Run all tasks concurrently
|
||||
await asyncio.gather(*tasks)
|
||||
|
||||
|
||||
def decrypt_process(runtimes: Dict[str, RuntimeInfo], sequences: List[Tuple[str, bytes]], args):
|
||||
asyncio.run(decrypt_process_async(runtimes, sequences, args))
|
||||
|
||||
|
||||
def get_platform_executable(args) -> str:
|
||||
"""
|
||||
Get the appropriate executable for the current platform
|
||||
"""
|
||||
logger = logging.getLogger('shot')
|
||||
|
||||
# If a specific executable is provided, use it
|
||||
if args.executable:
|
||||
if os.path.exists(args.executable):
|
||||
logger.info(f'{Fore.GREEN}Using specified executable: {args.executable}{Style.RESET_ALL}')
|
||||
return args.executable
|
||||
else:
|
||||
logger.warning(f'{Fore.YELLOW}Specified executable not found: {args.executable}{Style.RESET_ALL}')
|
||||
|
||||
helpers_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
system = platform.system().lower()
|
||||
machine = platform.machine().lower()
|
||||
|
||||
# Check for architecture-specific executables
|
||||
arch_specific_exe = f'pyarmor-1shot-{system}-{machine}'
|
||||
if system == 'windows':
|
||||
arch_specific_exe += '.exe'
|
||||
|
||||
arch_exe_path = os.path.join(helpers_dir, arch_specific_exe)
|
||||
if os.path.exists(arch_exe_path):
|
||||
logger.info(f'{Fore.GREEN}Using architecture-specific executable: {arch_specific_exe}{Style.RESET_ALL}')
|
||||
return arch_exe_path
|
||||
|
||||
platform_map = {
|
||||
'windows': 'pyarmor-1shot.exe',
|
||||
'linux': 'pyarmor-1shot',
|
||||
'darwin': 'pyarmor-1shot',
|
||||
}
|
||||
base_exe_name = platform_map.get(system, 'pyarmor-1shot')
|
||||
|
||||
# Then check for platform-specific executable
|
||||
platform_exe_path = os.path.join(helpers_dir, base_exe_name)
|
||||
if os.path.exists(platform_exe_path):
|
||||
logger.info(f'{Fore.GREEN}Using platform-specific executable: {base_exe_name}{Style.RESET_ALL}')
|
||||
return platform_exe_path
|
||||
|
||||
# Finally, check for generic executable
|
||||
generic_exe_path = os.path.join(helpers_dir, 'pyarmor-1shot')
|
||||
if os.path.exists(generic_exe_path):
|
||||
logger.info(f'{Fore.GREEN}Using generic executable: pyarmor-1shot{Style.RESET_ALL}')
|
||||
return generic_exe_path
|
||||
|
||||
logger.critical(f'{Fore.RED}Executable {base_exe_name} not found, please build it first or download on https://github.com/Lil-House/Pyarmor-Static-Unpack-1shot/releases {Style.RESET_ALL}')
|
||||
exit(1)
|
||||
|
||||
|
||||
def parse_args():
|
||||
@@ -94,6 +273,7 @@ def parse_args():
|
||||
'directory',
|
||||
help='the "root" directory of obfuscated scripts',
|
||||
type=str,
|
||||
nargs='?',
|
||||
)
|
||||
parser.add_argument(
|
||||
'-r',
|
||||
@@ -127,9 +307,61 @@ def parse_args():
|
||||
help='show pycdc stack related warnings',
|
||||
action='store_true',
|
||||
)
|
||||
parser.add_argument(
|
||||
'--menu',
|
||||
help='show interactive menu to select folder to unpack',
|
||||
action='store_true',
|
||||
)
|
||||
parser.add_argument(
|
||||
'--concurrent',
|
||||
help='number of concurrent deobfuscation processes (default: 4)',
|
||||
type=int,
|
||||
default=4,
|
||||
)
|
||||
parser.add_argument(
|
||||
'-e',
|
||||
'--executable',
|
||||
help='path to the pyarmor-1shot executable to use',
|
||||
type=str,
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def display_menu():
|
||||
to_unpack_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'to_unpack')
|
||||
|
||||
if not os.path.exists(to_unpack_dir):
|
||||
os.makedirs(to_unpack_dir)
|
||||
|
||||
folders = [d for d in os.listdir(to_unpack_dir)
|
||||
if os.path.isdir(os.path.join(to_unpack_dir, d))]
|
||||
|
||||
if not folders:
|
||||
print(f"{Fore.YELLOW}No folders found in {to_unpack_dir}{Style.RESET_ALL}")
|
||||
return None
|
||||
|
||||
print(f"\n{Fore.CYAN}=== Available Folders to Unpack ==={Style.RESET_ALL}")
|
||||
for i, folder in enumerate(folders, 1):
|
||||
print(f"{Fore.GREEN}[{i}]{Style.RESET_ALL} {folder}")
|
||||
|
||||
while True:
|
||||
try:
|
||||
choice = input(f"\n{Fore.YELLOW}Enter the number of the folder to unpack (or 'q' to quit): {Style.RESET_ALL}")
|
||||
if choice.lower() == 'q':
|
||||
return None
|
||||
|
||||
choice_idx = int(choice) - 1
|
||||
if 0 <= choice_idx < len(folders):
|
||||
selected_folder = folders[choice_idx]
|
||||
full_path = os.path.join(to_unpack_dir, selected_folder)
|
||||
print(f"{Fore.GREEN}Selected: {selected_folder}{Style.RESET_ALL}")
|
||||
return full_path
|
||||
else:
|
||||
print(f"{Fore.RED}Invalid choice. Please enter a number between 1 and {len(folders)}{Style.RESET_ALL}")
|
||||
except ValueError:
|
||||
print(f"{Fore.RED}Please enter a valid number{Style.RESET_ALL}")
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
logging.basicConfig(
|
||||
@@ -138,7 +370,7 @@ def main():
|
||||
)
|
||||
logger = logging.getLogger('shot')
|
||||
|
||||
print(r'''
|
||||
print(Fore.CYAN + r'''
|
||||
____ ____
|
||||
( __ ) ( __ )
|
||||
| |~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~| |
|
||||
@@ -153,10 +385,20 @@ def main():
|
||||
|
||||
For technology exchange only. Use at your own risk.
|
||||
GitHub: https://github.com/Lil-House/Pyarmor-Static-Unpack-1shot
|
||||
''')
|
||||
''' + Style.RESET_ALL)
|
||||
|
||||
# If menu option is selected or no directory is provided, show the menu
|
||||
if args.menu or not args.directory:
|
||||
selected_dir = display_menu()
|
||||
if selected_dir:
|
||||
args.directory = selected_dir
|
||||
else:
|
||||
print(f"{Fore.YELLOW}No directory selected. Exiting.{Style.RESET_ALL}")
|
||||
return
|
||||
|
||||
if args.runtime:
|
||||
specified_runtime = RuntimeInfo(args.runtime)
|
||||
print(specified_runtime)
|
||||
runtimes = {specified_runtime.serial_number: specified_runtime}
|
||||
else:
|
||||
specified_runtime = None
|
||||
@@ -167,71 +409,71 @@ def main():
|
||||
if args.output_dir and not os.path.exists(args.output_dir):
|
||||
os.makedirs(args.output_dir)
|
||||
|
||||
if os.path.isfile(args.directory):
|
||||
if specified_runtime is None:
|
||||
logger.error(f'{Fore.RED}Please specify `pyarmor_runtime` file by `-r` if input is a file{Style.RESET_ALL}')
|
||||
return
|
||||
logger.info(f'{Fore.CYAN}Single file mode{Style.RESET_ALL}')
|
||||
result = detect_process(args.directory, args.directory)
|
||||
if result is None:
|
||||
logger.error(f'{Fore.RED}No armored data found{Style.RESET_ALL}')
|
||||
return
|
||||
sequences.extend(result)
|
||||
decrypt_process(runtimes, sequences, args)
|
||||
return # single file mode ends here
|
||||
|
||||
dir_path: str
|
||||
dirs: List[str]
|
||||
files: List[str]
|
||||
for dir_path, dirs, files in os.walk(args.directory, followlinks=False):
|
||||
if '.no1shot' in files:
|
||||
logger.info(f'{Fore.YELLOW}Skipping {dir_path} because of `.no1shot`{Style.RESET_ALL}')
|
||||
dirs.clear()
|
||||
files.clear()
|
||||
continue
|
||||
for d in ['__pycache__', 'site-packages']:
|
||||
if d in dirs:
|
||||
dirs.remove(d)
|
||||
for file_name in files:
|
||||
if '.1shot.' in file_name:
|
||||
continue
|
||||
handled = False
|
||||
|
||||
file_path = os.path.join(dir_path, file_name)
|
||||
relative_path = os.path.relpath(file_path, args.directory)
|
||||
|
||||
if file_name.endswith('.pyz'):
|
||||
with open(file_path, 'rb') as f:
|
||||
head = f.read(16 * 1024 * 1024)
|
||||
if b'PY00' in head \
|
||||
and (not os.path.exists(file_path + '_extracted')
|
||||
or len(os.listdir(file_path + '_extracted')) == 0):
|
||||
logger.error(
|
||||
f'{Fore.RED}A PYZ file containing armored data is detected, but the PYZ file has not been extracted by other tools. This error is not a problem with this tool. If the folder is extracted by Pyinstxtractor, please read the output information of Pyinstxtractor carefully. ({relative_path}){Style.RESET_ALL}')
|
||||
continue
|
||||
|
||||
# is pyarmor_runtime?
|
||||
if not handled \
|
||||
and specified_runtime is None \
|
||||
if specified_runtime is None \
|
||||
and file_name.startswith('pyarmor_runtime') \
|
||||
and file_name.endswith(('.pyd', '.so', '.dylib')):
|
||||
try:
|
||||
new_runtime = RuntimeInfo(file_path)
|
||||
runtimes[new_runtime.serial_number] = new_runtime
|
||||
logger.info(
|
||||
f'Found new runtime: {new_runtime.serial_number} ({file_path})')
|
||||
f'{Fore.GREEN}Found new runtime: {new_runtime.serial_number} ({file_path}){Style.RESET_ALL}')
|
||||
print(new_runtime)
|
||||
handled = True
|
||||
continue
|
||||
except:
|
||||
pass
|
||||
|
||||
try:
|
||||
with open(file_path, 'rb') as f:
|
||||
beacon = f.read(16 * 1024 * 1024)
|
||||
except:
|
||||
logger.error(f'Failed to read file: {relative_path}')
|
||||
continue
|
||||
|
||||
# is UTF-8 source?
|
||||
# TODO: only support natural one line now
|
||||
if not handled and b'__pyarmor__(__name__, __file__,' in beacon:
|
||||
try:
|
||||
with open(file_path, 'r') as f:
|
||||
for line in f:
|
||||
if line.startswith('__pyarmor__(') and line.rstrip().endswith(')'):
|
||||
co = compile(line, '<str>', 'exec')
|
||||
bytes_raw = co.co_consts[0]
|
||||
assert type(bytes_raw) is bytes
|
||||
assert bytes_raw.startswith(b'PY')
|
||||
assert len(bytes_raw) > 64
|
||||
break
|
||||
logger.info(f'Found data in source: {relative_path}')
|
||||
# FIXME: bytes_raw can be kept from last iteration
|
||||
sequences.append((relative_path, bytes_raw))
|
||||
del bytes_raw
|
||||
handled = True
|
||||
except Exception as e:
|
||||
logger.error(f'Assume source, but {e} ({file_path})')
|
||||
|
||||
# TODO: is Nuitka package?
|
||||
# TODO: is pyc or single marshalled binary?
|
||||
result = detect_process(file_path, relative_path)
|
||||
if result is not None:
|
||||
sequences.extend(result)
|
||||
|
||||
if not runtimes:
|
||||
logger.error('No runtime found')
|
||||
logger.error(f'{Fore.RED}No runtime found{Style.RESET_ALL}')
|
||||
return
|
||||
if not sequences:
|
||||
logger.error('No armored data found')
|
||||
logger.error(f'{Fore.RED}No armored data found{Style.RESET_ALL}')
|
||||
return
|
||||
decrypt_process(runtimes, sequences, args)
|
||||
|
||||
|
@@ -86,6 +86,9 @@ int main(int argc, char* argv[])
|
||||
return 1;
|
||||
}
|
||||
|
||||
das_out_file.flush();
|
||||
das_out_file.close();
|
||||
|
||||
dc_out_file << "# Source Generated with Decompyle++\n";
|
||||
formatted_print(dc_out_file, "# File: %s (Python %d.%d%s)\n\n", dispname,
|
||||
mod.majorVer(), mod.minorVer(),
|
||||
@@ -97,5 +100,8 @@ int main(int argc, char* argv[])
|
||||
return 1;
|
||||
}
|
||||
|
||||
dc_out_file.flush();
|
||||
dc_out_file.close();
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
BIN
tests/compiled/binary_slice.3.12.pyc
Normal file
BIN
tests/compiled/binary_slice.3.12.pyc
Normal file
Binary file not shown.
BIN
tests/compiled/store_slice.3.12.pyc
Normal file
BIN
tests/compiled/store_slice.3.12.pyc
Normal file
Binary file not shown.
BIN
tests/compiled/test_loops3.3.12.pyc
Normal file
BIN
tests/compiled/test_loops3.3.12.pyc
Normal file
Binary file not shown.
BIN
tests/compiled/test_raise_varargs.3.12.pyc
Normal file
BIN
tests/compiled/test_raise_varargs.3.12.pyc
Normal file
Binary file not shown.
7
tests/input/binary_slice.py
Normal file
7
tests/input/binary_slice.py
Normal file
@@ -0,0 +1,7 @@
|
||||
l = [1,2,3,4,5,6]
|
||||
print(l[1:3])
|
||||
print(l[:2])
|
||||
print(l[3:])
|
||||
print(l[-4:])
|
||||
print(l[:-2])
|
||||
print(l[:])
|
8
tests/input/store_slice.py
Normal file
8
tests/input/store_slice.py
Normal file
@@ -0,0 +1,8 @@
|
||||
a = [0] * 16
|
||||
l = [1,2,3]
|
||||
a[13:] = l
|
||||
a[5] = 10
|
||||
a[:2] = [1,2]
|
||||
a[:] = range(16)
|
||||
|
||||
print(a)
|
34
tests/input/test_loops3.py
Normal file
34
tests/input/test_loops3.py
Normal file
@@ -0,0 +1,34 @@
|
||||
def loop1():
|
||||
iterable = [1, 2, 3]
|
||||
for item in iterable:
|
||||
pass
|
||||
|
||||
loop1()
|
||||
|
||||
def loop2():
|
||||
for i in range(2):
|
||||
print(i)
|
||||
|
||||
loop2()
|
||||
|
||||
def loop3():
|
||||
def loop():
|
||||
x = (1,2,3)
|
||||
l = []
|
||||
for i in x:
|
||||
l.append(i)
|
||||
return l
|
||||
|
||||
return loop()
|
||||
|
||||
loop3()
|
||||
|
||||
def loop4():
|
||||
for i in range(3):
|
||||
for j in range(2):
|
||||
print(i*j)
|
||||
|
||||
loop4()
|
||||
|
||||
for j in [1,2,3][::-1]:
|
||||
print("hi", j)
|
7
tests/input/test_raise_varargs.py
Normal file
7
tests/input/test_raise_varargs.py
Normal file
@@ -0,0 +1,7 @@
|
||||
import struct
|
||||
|
||||
def bytes_to_words(b):
|
||||
'''Convert a byte string (little-endian) to a list of 32-bit words.'''
|
||||
if len(b) % 4 != 0:
|
||||
raise ValueError('Input bytes length must be a multiple of 4 for word conversion.')
|
||||
return struct.unpack('<' + 'I' * (len(b) // 4), b)
|
7
tests/tokenized/binary_slice.txt
Normal file
7
tests/tokenized/binary_slice.txt
Normal file
@@ -0,0 +1,7 @@
|
||||
l = [ 1 , 2 , 3 , 4 , 5 , 6 ] <EOL>
|
||||
print ( l [ 1 : 3 ] ) <EOL>
|
||||
print ( l [ : 2 ] ) <EOL>
|
||||
print ( l [ 3 : ] ) <EOL>
|
||||
print ( l [ - 4 : ] ) <EOL>
|
||||
print ( l [ : - 2 ] ) <EOL>
|
||||
print ( l [ : ] ) <EOL>
|
7
tests/tokenized/store_slice.txt
Normal file
7
tests/tokenized/store_slice.txt
Normal file
@@ -0,0 +1,7 @@
|
||||
a = [ 0 ] * 16 <EOL>
|
||||
l = [ 1 , 2 , 3 ] <EOL>
|
||||
a [ 13 : ] = l <EOL>
|
||||
a [ 5 ] = 10 <EOL>
|
||||
a [ : 2 ] = [ 1 , 2 ] <EOL>
|
||||
a [ : ] = range ( 16 ) <EOL>
|
||||
print ( a ) <EOL>
|
46
tests/tokenized/test_loops3.txt
Normal file
46
tests/tokenized/test_loops3.txt
Normal file
@@ -0,0 +1,46 @@
|
||||
def loop1 ( ) : <EOL>
|
||||
<INDENT>
|
||||
iterable = [ 1 , 2 , 3 ] <EOL>
|
||||
for item in iterable : <EOL>
|
||||
<INDENT>
|
||||
pass <EOL>
|
||||
<OUTDENT>
|
||||
<OUTDENT>
|
||||
loop1 ( ) <EOL>
|
||||
def loop2 ( ) : <EOL>
|
||||
<INDENT>
|
||||
for i in range ( 2 ) : <EOL>
|
||||
<INDENT>
|
||||
print ( i ) <EOL>
|
||||
<OUTDENT>
|
||||
<OUTDENT>
|
||||
loop2 ( ) <EOL>
|
||||
def loop3 ( ) : <EOL>
|
||||
<INDENT>
|
||||
def loop ( ) : <EOL>
|
||||
<INDENT>
|
||||
x = ( 1 , 2 , 3 ) <EOL>
|
||||
l = [ ] <EOL>
|
||||
for i in x : <EOL>
|
||||
<INDENT>
|
||||
l . append ( i ) <EOL>
|
||||
<OUTDENT>
|
||||
return l <EOL>
|
||||
<OUTDENT>
|
||||
return loop ( ) <EOL>
|
||||
<OUTDENT>
|
||||
loop3 ( ) <EOL>
|
||||
def loop4 ( ) : <EOL>
|
||||
<INDENT>
|
||||
for i in range ( 3 ) : <EOL>
|
||||
<INDENT>
|
||||
for j in range ( 2 ) : <EOL>
|
||||
<INDENT>
|
||||
print ( i * j ) <EOL>
|
||||
<OUTDENT>
|
||||
<OUTDENT>
|
||||
<OUTDENT>
|
||||
loop4 ( ) <EOL>
|
||||
for j in [ 1 , 2 , 3 ] [ : : - 1 ] : <EOL>
|
||||
<INDENT>
|
||||
print ( 'hi' , j ) <EOL>
|
9
tests/tokenized/test_raise_varargs.txt
Normal file
9
tests/tokenized/test_raise_varargs.txt
Normal file
@@ -0,0 +1,9 @@
|
||||
import struct <EOL>
|
||||
def bytes_to_words ( b ) : <EOL>
|
||||
<INDENT>
|
||||
'Convert a byte string (little-endian) to a list of 32-bit words.' <EOL>
|
||||
if len ( b ) % 4 != 0 : <EOL>
|
||||
<INDENT>
|
||||
raise ValueError ( 'Input bytes length must be a multiple of 4 for word conversion.' ) <EOL>
|
||||
<OUTDENT>
|
||||
return struct . unpack ( '<' + 'I' * ( len ( b ) // 4 ) , b ) <EOL>
|
Reference in New Issue
Block a user