Compare commits
49 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
682b47fbd3 | ||
|
|
9d82c727b6 | ||
|
|
c425d86ed6 | ||
|
|
3ed06686ab | ||
|
|
7e4fb0a8c4 | ||
|
|
ebfe0c48a0 | ||
|
|
45d854e590 | ||
|
|
18d79c8e04 | ||
|
|
92c38344e3 | ||
|
|
b596f42b7e | ||
|
|
86c5051fcf | ||
|
|
f838f8ee94 | ||
|
|
fb7a93732d | ||
|
|
181763070c | ||
|
|
c3a491ac46 | ||
|
|
641c55a6a2 | ||
|
|
432675f2e6 | ||
|
|
635c87dabd | ||
|
|
6b6c33e545 | ||
|
|
7c5ba2fdbe | ||
|
|
692335ee62 | ||
|
|
c63256c628 | ||
|
|
66659da66d | ||
|
|
32d60b5321 | ||
|
|
9e9f480075 | ||
|
|
7fc2a1a625 | ||
|
|
8aea956e77 | ||
|
|
ac58fff9ce | ||
|
|
7a59d1acd1 | ||
|
|
5c6b9611d8 | ||
|
|
dc034b1d55 | ||
|
|
5d988e01fc | ||
|
|
85cc1454ea | ||
|
|
f5ef4bbc03 | ||
|
|
8d7a9235b8 | ||
|
|
70f386193a | ||
|
|
d37e573d9e | ||
|
|
e850b5495a | ||
|
|
9390edeb79 | ||
|
|
d11beb10af | ||
|
|
4c124a33c0 | ||
|
|
5b38c532a2 | ||
|
|
f7fae450a0 | ||
|
|
1ff6e721c7 | ||
|
|
487def2b45 | ||
|
|
c27606d442 | ||
|
|
2381893b72 | ||
|
|
03b4fcc2ea | ||
|
|
587dec6b6a |
3
.gitattributes
vendored
Normal file
3
.gitattributes
vendored
Normal file
@@ -0,0 +1,3 @@
|
||||
.gitattributes export-ignore
|
||||
.gitignore export-ignore
|
||||
README.md export-ignore
|
||||
133
.gitignore
vendored
133
.gitignore
vendored
@@ -1,2 +1,133 @@
|
||||
*.pyc
|
||||
# Gedit/other editors
|
||||
*.bak
|
||||
*~
|
||||
|
||||
# Vim.gitignore
|
||||
# https://github.com/github/gitignore/blob/master/Global/Vim.gitignore
|
||||
# Swap
|
||||
[._]*.s[a-v][a-z]
|
||||
[._]*.sw[a-p]
|
||||
[._]s[a-v][a-z]
|
||||
[._]sw[a-p]
|
||||
|
||||
# Session
|
||||
Session.vim
|
||||
|
||||
# Temporary
|
||||
.netrwhist
|
||||
*~
|
||||
# Auto-generated tag files
|
||||
tags
|
||||
# <>
|
||||
|
||||
# Python.gitignore
|
||||
# https://github.com/github/gitignore/blob/master/Python.gitignore
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
.hypothesis/
|
||||
cover/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
.static_storage/
|
||||
.media/
|
||||
local_settings.py
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# pyenv
|
||||
.python-version
|
||||
|
||||
# celery beat schedule file
|
||||
celerybeat-schedule
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
.env
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
# <>
|
||||
|
||||
# PyCharm project settings
|
||||
.idea
|
||||
|
||||
18
LICENSE
18
LICENSE
@@ -1,17 +1,7 @@
|
||||
DWTFYWWI LICENSE
|
||||
Version 1, January 2006
|
||||
Copyright 2017 © R4SAS <r4sas@i2pmail.org>
|
||||
|
||||
Copyright (C) 2017-2018 R4SAS <r4sas@i2pmail.org>
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
|
||||
|
||||
Preamble
|
||||
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
|
||||
|
||||
The licenses for most software are designed to take away your
|
||||
freedom to share and change it. By contrast, the DWTFYWWI or Do
|
||||
Whatever The Fuck You Want With It license is intended to guarantee
|
||||
your freedom to share and change the software--to make sure the
|
||||
software is free for all its users.
|
||||
|
||||
DWTFYWWI LICENSE
|
||||
TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
|
||||
0. The author grants everyone permission to do whatever the fuck they
|
||||
want with the software, whatever the fuck that may be.
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
|
||||
46
README.md
46
README.md
@@ -1,55 +1,67 @@
|
||||
[](https://github.com/r4sas/PBinCLI/blob/master/LICENSE)
|
||||
[](https://github.com/r4sas/PBinCLI/tags/)
|
||||
[](https://www.codacy.com/app/r4sas/PBinCLI?utm_source=github.com&utm_medium=referral&utm_content=r4sas/PBinCLI&utm_campaign=Badge_Grade)
|
||||
|
||||
PBinCLI
|
||||
=====
|
||||
|
||||
#### [PrivateBin](https://github.com/PrivateBin/PrivateBin/) CLI
|
||||
PBinCLI is command line client for [PrivateBin](https://github.com/PrivateBin/PrivateBin/) written on Python 3.
|
||||
|
||||
Installing
|
||||
-----
|
||||
```bash
|
||||
$ virtualenv --python=python3 venv
|
||||
$ . venv/bin/activate
|
||||
$ pip install pbincli
|
||||
virtualenv --python=python3 venv
|
||||
. venv/bin/activate
|
||||
pip install pbincli
|
||||
```
|
||||
|
||||
Usage
|
||||
-----
|
||||
By default pbincli configured to use https://paste.i2pd.xyz/ for sending and receiving pastes. No proxy used by default.
|
||||
By default pbincli configured to use `https://paste.i2pd.xyz/` for sending and receiving pastes. No proxy used by default.
|
||||
|
||||
You can create config file with variables `server` and `proxy` in `~/.config/pbincli/pbincli.conf` to use different settings.
|
||||
|
||||
Example contents:
|
||||
|
||||
```
|
||||
```ini
|
||||
server=https://paste.i2pd.xyz/
|
||||
proxy=http://127.0.0.1:3128
|
||||
```
|
||||
|
||||
Run inside `venv` command:
|
||||
|
||||
$ pbincli send --text "Hello!"
|
||||
```bash
|
||||
pbincli send --text "Hello!"
|
||||
```
|
||||
|
||||
Or use stdin input to read text for paste:
|
||||
|
||||
$ pbincli send - <<EOF
|
||||
Hello! This is test paste!
|
||||
EOF
|
||||
```bash
|
||||
pbincli send - <<EOF
|
||||
Hello! This is test paste!
|
||||
EOF
|
||||
```
|
||||
|
||||
It will send string `Hello! This is test paste!` to PrivateBin.
|
||||
|
||||
To send file use `--file` or `-f` with filename. Example:
|
||||
|
||||
$ pbincli send -c "My document" -f info.pdf
|
||||
|
||||
```bash
|
||||
pbincli send -c "My document" -f info.pdf
|
||||
```
|
||||
|
||||
To retrieve paste from server, use `get` command with paste info.
|
||||
|
||||
It must be formated like `pasteID#passphrase`. Example:
|
||||
|
||||
$ pbincli get 49eeb1326cfa9491#vfeortoVWaYeJlviDdhxQBtj5e0I2kArpynrtu/tnGs=
|
||||
|
||||
```bash
|
||||
pbincli get 49eeb1326cfa9491#vfeortoVWaYeJlviDdhxQBtj5e0I2kArpynrtu/tnGs=
|
||||
```
|
||||
More info you can find by typing
|
||||
|
||||
$ pbincli [-h] {send, get, delete}
|
||||
```bash
|
||||
pbincli [-h] {send, get, delete}
|
||||
```
|
||||
|
||||
TODO
|
||||
----
|
||||
@@ -57,5 +69,5 @@ Write a more complete usage documentation.
|
||||
|
||||
License
|
||||
-------
|
||||
This project is licensed under the DWTFYWWI license, which can be found in the file
|
||||
[LICENSE](LICENSE) in the root of the project source code.
|
||||
This project is licensed under the MIT license, which can be found in the file
|
||||
[LICENSE](https://github.com/r4sas/PBinCLI/blob/master/LICENSE) in the root of the project source code.
|
||||
|
||||
91
README.rst
Normal file
91
README.rst
Normal file
@@ -0,0 +1,91 @@
|
||||
|
||||
|
||||
.. image:: https://img.shields.io/github/license/r4sas/PBinCLI.svg
|
||||
:target: https://github.com/r4sas/PBinCLI/blob/master/LICENSE
|
||||
:alt: GitHub license
|
||||
|
||||
|
||||
.. image:: https://img.shields.io/github/tag/r4sas/PBinCLI.svg
|
||||
:target: https://github.com/r4sas/PBinCLI/tags/
|
||||
:alt: GitHub tag
|
||||
|
||||
|
||||
.. image:: https://api.codacy.com/project/badge/Grade/4f24f43356a84621bbd9078c4b3f1b70
|
||||
:target: https://www.codacy.com/app/r4sas/PBinCLI?utm_source=github.com&utm_medium=referral&utm_content=r4sas/PBinCLI&utm_campaign=Badge_Grade
|
||||
:alt: Codacy Badge
|
||||
|
||||
|
||||
PBinCLI
|
||||
=======
|
||||
|
||||
PBinCLI is command line client for `PrivateBin <https://github.com/PrivateBin/PrivateBin/>`_ written on Python 3.
|
||||
|
||||
Installing
|
||||
----------
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
virtualenv --python=python3 venv
|
||||
. venv/bin/activate
|
||||
pip install pbincli
|
||||
|
||||
Usage
|
||||
-----
|
||||
|
||||
By default pbincli configured to use ``https://paste.i2pd.xyz/`` for sending and receiving pastes. No proxy used by default.
|
||||
|
||||
You can create config file with variables ``server`` and ``proxy`` in ``~/.config/pbincli/pbincli.conf`` to use different settings.
|
||||
|
||||
Example contents:
|
||||
|
||||
.. code-block:: ini
|
||||
|
||||
server=https://paste.i2pd.xyz/
|
||||
proxy=http://127.0.0.1:3128
|
||||
|
||||
Run inside ``venv`` command:
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
pbincli send --text "Hello!"
|
||||
|
||||
Or use stdin input to read text for paste:
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
pbincli send - <<EOF
|
||||
Hello! This is test paste!
|
||||
EOF
|
||||
|
||||
It will send string ``Hello! This is test paste!`` to PrivateBin.
|
||||
|
||||
To send file use ``--file`` or ``-f`` with filename. Example:
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
pbincli send -c "My document" -f info.pdf
|
||||
|
||||
To retrieve paste from server, use ``get`` command with paste info.
|
||||
|
||||
It must be formated like ``pasteID#passphrase``. Example:
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
pbincli get 49eeb1326cfa9491#vfeortoVWaYeJlviDdhxQBtj5e0I2kArpynrtu/tnGs=
|
||||
|
||||
More info you can find by typing
|
||||
|
||||
.. code-block:: bash
|
||||
|
||||
pbincli [-h] {send, get, delete}
|
||||
|
||||
TODO
|
||||
----
|
||||
|
||||
Write a more complete usage documentation.
|
||||
|
||||
License
|
||||
-------
|
||||
|
||||
This project is licensed under the MIT license, which can be found in the file
|
||||
`LICENSE <https://github.com/r4sas/PBinCLI/blob/master/LICENSE>`_ in the root of the project source code.
|
||||
@@ -2,6 +2,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
__author__ = "R4SAS <r4sas@i2pmail.org>"
|
||||
__version__ = "0.1"
|
||||
__version__ = "0.3.0"
|
||||
__copyright__ = "Copyright (c) R4SAS"
|
||||
__license__ = "DWTFYWWI"
|
||||
__license__ = "MIT"
|
||||
|
||||
@@ -1,219 +1,162 @@
|
||||
import json, hashlib, ntpath, os, sys, zlib
|
||||
import pbincli.actions
|
||||
from sjcl import SJCL
|
||||
from pbincli.format import Paste
|
||||
from pbincli.utils import PBinCLIError
|
||||
|
||||
from base64 import b64encode, b64decode
|
||||
from mimetypes import guess_type
|
||||
from pbincli.utils import PBinCLIException, check_readable, check_writable, json_load_byteified
|
||||
def send(args, api_client, settings=None):
|
||||
from pbincli.api import Shortener
|
||||
if args.short:
|
||||
shortener = Shortener(settings)
|
||||
|
||||
|
||||
def path_leaf(path):
|
||||
head, tail = ntpath.split(path)
|
||||
return tail or ntpath.basename(head)
|
||||
|
||||
|
||||
def decompress(s):
|
||||
return zlib.decompress(bytearray(map(ord, b64decode(s.encode('utf-8')).decode('utf-8'))), -zlib.MAX_WBITS)
|
||||
|
||||
def compress(s):
|
||||
co = zlib.compressobj(wbits=-zlib.MAX_WBITS)
|
||||
b = co.compress(s) + co.flush()
|
||||
|
||||
return b64encode(''.join(map(chr, b)).encode('utf-8'))
|
||||
|
||||
def send(args, api_client):
|
||||
if args.stdin:
|
||||
text = args.stdin.read()
|
||||
elif args.text:
|
||||
text = args.text
|
||||
elif args.file:
|
||||
text = "Sending a file to you!"
|
||||
if not args.notext:
|
||||
if args.text:
|
||||
text = args.text
|
||||
elif args.stdin:
|
||||
text = args.stdin.read()
|
||||
elif not args.file:
|
||||
PBinCLIError("Nothing to send!")
|
||||
else:
|
||||
print("Nothing to send!")
|
||||
sys.exit(1)
|
||||
text = ""
|
||||
|
||||
# Formatting request
|
||||
request = {'expire':args.expire,'formatter':args.format,'burnafterreading':int(args.burn),'opendiscussion':int(args.discus)}
|
||||
paste = Paste(args.debug)
|
||||
|
||||
passphrase = b64encode(os.urandom(32))
|
||||
if args.debug: print("Passphrase:\t{}".format(passphrase))
|
||||
# get from server supported paste format version and update object
|
||||
version = api_client.getVersion()
|
||||
paste.setVersion(version)
|
||||
|
||||
# set compression type, works only on v2 pastes
|
||||
if version == 2:
|
||||
paste.setCompression(args.compression)
|
||||
|
||||
# add text in paste (if it provided)
|
||||
paste.setText(text)
|
||||
|
||||
# If we set PASSWORD variable
|
||||
if args.password:
|
||||
digest = hashlib.sha256(args.password.encode("UTF-8")).hexdigest()
|
||||
password = passphrase + digest.encode("UTF-8")
|
||||
else:
|
||||
password = passphrase
|
||||
|
||||
if args.debug: print("Password:\t{}".format(password))
|
||||
|
||||
# Encrypting text
|
||||
cipher = SJCL().encrypt(compress(text.encode('utf-8')), password, mode='gcm')
|
||||
|
||||
# TODO: should be implemented in upstream
|
||||
for k in ['salt', 'iv', 'ct']: cipher[k] = cipher[k].decode()
|
||||
|
||||
request['data'] = json.dumps(cipher, ensure_ascii=False).replace(' ','')
|
||||
paste.setPassword(args.password)
|
||||
|
||||
# If we set FILE variable
|
||||
if args.file:
|
||||
check_readable(args.file)
|
||||
with open(args.file, "rb") as f:
|
||||
contents = f.read()
|
||||
f.close()
|
||||
mime = guess_type(args.file)
|
||||
if args.debug: print("Filename:\t{}\nMIME-type:\t{}".format(path_leaf(args.file), mime[0]))
|
||||
paste.setAttachment(args.file)
|
||||
|
||||
file = "data:" + mime[0] + ";base64," + b64encode(contents).decode()
|
||||
filename = path_leaf(args.file)
|
||||
paste.encrypt(
|
||||
formatter = args.format,
|
||||
burnafterreading = args.burn,
|
||||
discussion = args.discus,
|
||||
expiration = args.expire)
|
||||
|
||||
cipherfile = SJCL().encrypt(compress(file.encode('utf-8')), password, mode='gcm')
|
||||
# TODO: should be implemented in upstream
|
||||
for k in ['salt', 'iv', 'ct']: cipherfile[k] = cipherfile[k].decode()
|
||||
cipherfilename = SJCL().encrypt(compress(filename.encode('utf-8')), password, mode='gcm')
|
||||
for k in ['salt', 'iv', 'ct']: cipherfilename[k] = cipherfilename[k].decode()
|
||||
request = paste.getJSON()
|
||||
|
||||
request['attachment'] = json.dumps(cipherfile, ensure_ascii=False).replace(' ','')
|
||||
request['attachmentname'] = json.dumps(cipherfilename, ensure_ascii=False).replace(' ','')
|
||||
|
||||
if args.debug: print("Request:\t{}".format(request))
|
||||
if args.debug:
|
||||
print("Passphrase:\t{}".format(paste.getHash()))
|
||||
print("Request:\t{}".format(request))
|
||||
|
||||
# If we use dry option, exit now
|
||||
if args.dry: sys.exit(0)
|
||||
if args.dry: exit(0)
|
||||
|
||||
result = api_client.post(request)
|
||||
|
||||
if args.debug: print("Response:\t{}\n".format(result))
|
||||
|
||||
# Paste was sent. Checking for returned status code
|
||||
if not result['status']: # return code is zero
|
||||
passphrase = paste.getHash()
|
||||
|
||||
print("Paste uploaded!\nPasteID:\t{}\nPassword:\t{}\nDelete token:\t{}\n\nLink:\t\t{}?{}#{}".format(
|
||||
result['id'],
|
||||
passphrase,
|
||||
result['deletetoken'],
|
||||
settings['server'],
|
||||
result['id'],
|
||||
passphrase))
|
||||
elif result['status']: # return code is other then zero
|
||||
PBinCLIError("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
else: # or here no status field in response or it is empty
|
||||
PBinCLIError("Something went wrong...\nError: Empty response.")
|
||||
|
||||
if args.short:
|
||||
print("\nQuerying URL shortening service...")
|
||||
shortener.getlink("{}?{}#{}".format(
|
||||
settings['server'],
|
||||
result['id'],
|
||||
passphrase))
|
||||
|
||||
|
||||
def get(args, api_client, settings=None):
|
||||
from pbincli.utils import check_writable, json_encode
|
||||
|
||||
try:
|
||||
result = json.loads(result)
|
||||
except ValueError as e:
|
||||
print("PBinCLI Error: {}".format(e))
|
||||
sys.exit(1)
|
||||
pasteid, passphrase = args.pasteinfo.split("#")
|
||||
except ValueError:
|
||||
PBinCLIError("Provided info hasn't contain valid PasteID#Passphrase string")
|
||||
|
||||
if 'status' in result and not result['status']:
|
||||
print("Paste uploaded!\nPasteID:\t{}\nPassword:\t{}\nDelete token:\t{}\n\nLink:\t\t{}?{}#{}".format(result['id'], passphrase.decode(), result['deletetoken'], api_client.server, result['id'], passphrase.decode()))
|
||||
elif 'status' in result and result['status']:
|
||||
print("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("Something went wrong...\nError: Empty response.")
|
||||
sys.exit(1)
|
||||
if not (pasteid and passphrase):
|
||||
PBinCLIError("Incorrect request")
|
||||
|
||||
if args.debug: print("PasteID:\t{}\nPassphrase:\t{}".format(pasteid, passphrase))
|
||||
|
||||
def get(args, api_client):
|
||||
pasteid, passphrase = args.pasteinfo.split("#")
|
||||
paste = Paste(args.debug)
|
||||
|
||||
if pasteid and passphrase:
|
||||
if args.debug: print("PasteID:\t{}\nPassphrase:\t{}".format(pasteid, passphrase))
|
||||
if args.password:
|
||||
paste.setPassword(args.password)
|
||||
if args.debug: print("Password:\t{}".format(args.password))
|
||||
|
||||
if args.password:
|
||||
digest = hashlib.sha256(args.password.encode("UTF-8")).hexdigest()
|
||||
password = passphrase + digest.encode("UTF-8")
|
||||
else:
|
||||
password = passphrase
|
||||
|
||||
if args.debug: print("Password:\t{}".format(password))
|
||||
|
||||
result = api_client.get(pasteid)
|
||||
else:
|
||||
print("PBinCLI error: Incorrect request")
|
||||
sys.exit(1)
|
||||
result = api_client.get(pasteid)
|
||||
|
||||
if args.debug: print("Response:\t{}\n".format(result))
|
||||
|
||||
try:
|
||||
result = json.loads(result)
|
||||
except ValueError as e:
|
||||
print("PBinCLI Error: {}".format(e))
|
||||
sys.exit(1)
|
||||
# Paste was received. Checking received status code
|
||||
if not result['status']: # return code is zero
|
||||
print("Paste received!")
|
||||
|
||||
if 'status' in result and not result['status']:
|
||||
print("Paste received! Text inside:")
|
||||
data = json.loads(result['data'])
|
||||
version = result['v'] if 'v' in result else 1
|
||||
paste.setVersion(version)
|
||||
|
||||
if args.debug: print("Text:\t{}\n".format(data))
|
||||
if version == 2:
|
||||
if args.debug: print("Authentication data:\t{}".format(result['adata']))
|
||||
|
||||
text = SJCL().decrypt(data, password)
|
||||
print("{}\n".format(decompress(text.decode())))
|
||||
paste.setHash(passphrase)
|
||||
paste.loadJSON(result)
|
||||
paste.decrypt()
|
||||
|
||||
check_writable("paste.txt")
|
||||
with open("paste.txt", "wb") as f:
|
||||
f.write(decompress(text.decode()))
|
||||
f.close
|
||||
text = paste.getText()
|
||||
|
||||
if 'attachment' in result and 'attachmentname' in result:
|
||||
print("Found file, attached to paste. Decoding it and saving")
|
||||
if args.debug: print("Decoded text size: {}\n".format(len(text)))
|
||||
|
||||
cipherfile = json.loads(result['attachment'])
|
||||
cipherfilename = json.loads(result['attachmentname'])
|
||||
|
||||
if args.debug: print("Name:\t{}\nData:\t{}".format(cipherfilename, cipherfile))
|
||||
|
||||
attachmentf = SJCL().decrypt(cipherfile, password)
|
||||
attachmentname = SJCL().decrypt(cipherfilename, password)
|
||||
|
||||
attachment = decompress(attachmentf.decode('utf-8')).decode('utf-8').split(',', 1)[1]
|
||||
file = b64decode(attachment)
|
||||
filename = decompress(attachmentname.decode('utf-8')).decode('utf-8')
|
||||
|
||||
print("Filename:\t{}\n".format(filename))
|
||||
if len(text):
|
||||
if args.debug: print("{}\n".format(text.decode()))
|
||||
filename = "paste-" + pasteid + ".txt"
|
||||
print("Found text in paste. Saving it to {}".format(filename))
|
||||
|
||||
check_writable(filename)
|
||||
with open(filename, "wb") as f:
|
||||
f.write(file)
|
||||
f.close
|
||||
f.write(text)
|
||||
f.close()
|
||||
|
||||
if 'burnafterreading' in result['meta'] and result['meta']['burnafterreading']:
|
||||
attachment, attachment_name = paste.getAttachment()
|
||||
|
||||
if attachment:
|
||||
print("Found file, attached to paste. Saving it to {}\n".format(attachment_name))
|
||||
|
||||
check_writable(attachment_name)
|
||||
with open(attachment_name, "wb") as f:
|
||||
f.write(attachment)
|
||||
f.close()
|
||||
|
||||
if version == 1 and 'meta' in result and 'burnafterreading' in result['meta'] and result['meta']['burnafterreading']:
|
||||
print("Burn afrer reading flag found. Deleting paste...")
|
||||
result = api_client.delete(pasteid, 'burnafterreading')
|
||||
api_client.delete(json_encode({'pasteid':pasteid,'deletetoken':'burnafterreading'}))
|
||||
|
||||
if args.debug: print("Delete response:\t{}\n".format(result))
|
||||
|
||||
try:
|
||||
result = json.loads(result)
|
||||
except ValueError as e:
|
||||
print("PBinCLI Error: {}".format(e))
|
||||
sys.exit(1)
|
||||
|
||||
if 'status' in result and not result['status']:
|
||||
print("Paste successfully deleted!")
|
||||
elif 'status' in result and result['status']:
|
||||
print("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("Something went wrong...\nError: Empty response.")
|
||||
sys.exit(1)
|
||||
|
||||
elif 'status' in result and result['status']:
|
||||
print("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("Something went wrong...\nError: Empty response.")
|
||||
sys.exit(1)
|
||||
elif result['status']: # return code is other then zero
|
||||
PBinCLIError("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
else: # or here no status field in response or it is empty
|
||||
PBinCLIError("Something went wrong...\nError: Empty response.")
|
||||
|
||||
|
||||
def delete(args, api_client):
|
||||
def delete(args, api_client, settings=None):
|
||||
from pbincli.utils import json_encode
|
||||
|
||||
pasteid = args.paste
|
||||
token = args.token
|
||||
|
||||
if args.debug: print("PasteID:\t{}\nToken:\t\t{}".format(pasteid, token))
|
||||
|
||||
result = api_client.delete(pasteid, token)
|
||||
|
||||
if args.debug: print("Response:\t{}\n".format(result))
|
||||
|
||||
try:
|
||||
result = json.loads(result)
|
||||
except ValueError as e:
|
||||
print("PBinCLI Error: {}".format(e))
|
||||
sys.exit(1)
|
||||
|
||||
if 'status' in result and not result['status']:
|
||||
print("Paste successfully deleted!")
|
||||
elif 'status' in result and result['status']:
|
||||
print("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("Something went wrong...\nError: Empty response.")
|
||||
sys.exit(1)
|
||||
api_client.delete(json_encode({'pasteid':pasteid,'deletetoken':token}))
|
||||
|
||||
251
pbincli/api.py
251
pbincli/api.py
@@ -1,27 +1,248 @@
|
||||
import requests
|
||||
from requests import HTTPError
|
||||
from pbincli.utils import PBinCLIError
|
||||
|
||||
def _config_requests(settings=None):
|
||||
if settings['proxy']:
|
||||
proxy = {settings['proxy'].split('://')[0]: settings['proxy']}
|
||||
else:
|
||||
proxy = {}
|
||||
|
||||
if settings['no_insecure_warning']:
|
||||
from requests.packages.urllib3.exceptions import InsecureRequestWarning
|
||||
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
|
||||
|
||||
session = requests.Session()
|
||||
session.verify = not settings['no_check_certificate']
|
||||
|
||||
return session, proxy
|
||||
|
||||
|
||||
class PrivateBin:
|
||||
def __init__(self, server, proxy=None):
|
||||
self.server = server
|
||||
def __init__(self, settings=None):
|
||||
self.server = settings['server']
|
||||
self.headers = {'X-Requested-With': 'JSONHttpRequest'}
|
||||
if proxy:
|
||||
self.proxy = {proxy.split('://')[0]: proxy}
|
||||
else:
|
||||
self.proxy = {}
|
||||
|
||||
self.session, self.proxy = _config_requests(settings)
|
||||
|
||||
def post(self, request):
|
||||
r = requests.post(url = self.server, headers = self.headers, proxies = self.proxy, data = request)
|
||||
return r.text
|
||||
result = self.session.post(
|
||||
url = self.server,
|
||||
headers = self.headers,
|
||||
proxies = self.proxy,
|
||||
data = request)
|
||||
|
||||
try:
|
||||
return result.json()
|
||||
except ValueError:
|
||||
PBinCLIError("Unable parse response as json. Received (size = {}):\n{}".format(len(result.text), result.text))
|
||||
|
||||
|
||||
def get(self, request):
|
||||
url = self.server + "?" + request
|
||||
r = requests.get(url = url, headers = self.headers, proxies = self.proxy)
|
||||
return r.text
|
||||
return self.session.get(
|
||||
url = self.server + "?" + request,
|
||||
headers = self.headers,
|
||||
proxies = self.proxy).json()
|
||||
|
||||
|
||||
def delete(self, pasteid, token):
|
||||
request = {'pasteid':pasteid,'deletetoken':token}
|
||||
r = requests.post(url = self.server, headers = self.headers, proxies = self.proxy, data = request)
|
||||
return r.text
|
||||
def delete(self, request):
|
||||
# using try as workaround for versions < 1.3 due to we cant detect
|
||||
# if server used version 1.2, where auto-deletion is added
|
||||
try:
|
||||
result = self.session.post(
|
||||
url = self.server,
|
||||
headers = self.headers,
|
||||
proxies = self.proxy,
|
||||
data = request).json()
|
||||
except ValueError:
|
||||
# unable parse response as json because it can be empty (1.2), so simulate correct answer
|
||||
print("NOTICE: Received empty response. We interpret that as our paste has already been deleted.")
|
||||
from json import loads as json_loads
|
||||
result = json_loads('{"status":0}')
|
||||
|
||||
if not result['status']:
|
||||
print("Paste successfully deleted!")
|
||||
elif result['status']:
|
||||
PBinCLIError("Something went wrong...\nError:\t\t{}".format(result['message']))
|
||||
else:
|
||||
PBinCLIError("Something went wrong...\nError: Empty response.")
|
||||
|
||||
|
||||
def getVersion(self):
|
||||
jsonldSchema = self.session.get(
|
||||
url = self.server + '?jsonld=paste',
|
||||
proxies = self.proxy).json()
|
||||
return jsonldSchema['@context']['v']['@value'] \
|
||||
if ('@context' in jsonldSchema and
|
||||
'v' in jsonldSchema['@context'] and
|
||||
'@value' in jsonldSchema['@context']['v']) \
|
||||
else 1
|
||||
|
||||
class Shortener:
|
||||
"""Some parts of this class was taken from
|
||||
python-yourls (https://github.com/tflink/python-yourls/) library
|
||||
"""
|
||||
def __init__(self, settings=None):
|
||||
self.api = settings['short_api']
|
||||
|
||||
# we checking which service is used, because some services doesn't require
|
||||
# any authentication, or have only one domain on which it working
|
||||
if self.api == 'yourls':
|
||||
self._yourls_init(settings)
|
||||
elif self.api == 'isgd' or self.api == 'vgd':
|
||||
self._gd_init()
|
||||
|
||||
self.session, self.proxy = _config_requests(settings)
|
||||
|
||||
|
||||
def _yourls_init(self, settings):
|
||||
if not settings['short_url']:
|
||||
PBinCLIError("YOURLS: An API URL is required")
|
||||
|
||||
# setting API URL
|
||||
apiurl = settings['short_url']
|
||||
if apiurl.endswith('/yourls-api.php'):
|
||||
self.apiurl = apiurl
|
||||
elif apiurl.endswith('/'):
|
||||
self.apiurl = apiurl + 'yourls-api.php'
|
||||
else:
|
||||
PBinCLIError("YOURLS: Incorrect URL is provided.\n" +
|
||||
"It must contain full address to 'yourls-api.php' script (like https://example.com/yourls-api.php)\n" +
|
||||
"or just contain instance URL with '/' at the end (like https://example.com/)")
|
||||
|
||||
# validating for required credentials
|
||||
if settings['short_user'] and settings['short_pass'] and settings['short_token'] is None:
|
||||
self.auth_args = {'username': settings['short_user'], 'password': settings['short_pass']}
|
||||
elif settings['short_user'] is None and settings['short_pass'] is None and settings['short_token']:
|
||||
self.auth_args = {'signature': settings['short_token']}
|
||||
elif settings['short_user'] is None and settings['short_pass'] is None and settings['short_token'] is None:
|
||||
self.auth_args = {}
|
||||
else:
|
||||
PBinCLIError("YOURLS: either username and password or token are required. Otherwise set to default (None)")
|
||||
|
||||
|
||||
def _gd_init(self):
|
||||
if self.api == 'isgd':
|
||||
self.apiurl = 'https://is.gd/'
|
||||
else:
|
||||
self.apiurl = 'https://v.gd/'
|
||||
self.useragent = 'Mozilla/5.0 (compatible; pbincli - https://github.com/r4sas/pbincli/)'
|
||||
|
||||
|
||||
def getlink(self, url):
|
||||
# that is api -> function mapper for running service-related function when getlink() used
|
||||
servicesList = {
|
||||
'yourls': self._yourls,
|
||||
'clckru': self._clckru,
|
||||
'tinyurl': self._tinyurl,
|
||||
'isgd': self._gd,
|
||||
'vgd': self._gd,
|
||||
'cuttly': self._cuttly
|
||||
}
|
||||
# run function selected by choosen API
|
||||
servicesList[self.api](url)
|
||||
|
||||
|
||||
def _yourls(self,url):
|
||||
request = {'action': 'shorturl', 'format': 'json', 'url': url}
|
||||
request.update(self.auth_args)
|
||||
|
||||
result = self.session.post(
|
||||
url = self.apiurl,
|
||||
proxies = self.proxy,
|
||||
data = request)
|
||||
|
||||
try:
|
||||
result.raise_for_status()
|
||||
except HTTPError:
|
||||
try:
|
||||
response = result.json()
|
||||
except ValueError:
|
||||
PBinCLIError("YOURLS: Unable parse response. Received (size = {}):\n{}".format(len(result.text), result.text))
|
||||
else:
|
||||
PBinCLIError("YOURLS: Received error from API: {} with JSON {}".format(result, response))
|
||||
else:
|
||||
response = result.json()
|
||||
|
||||
if {'status', 'statusCode', 'message'} <= set(response.keys()):
|
||||
if response['status'] == 'fail':
|
||||
PBinCLIError("YOURLS: Received error from API: {}".format(response['message']))
|
||||
if not 'shorturl' in response:
|
||||
PBinCLIError("YOURLS: Unknown error: {}".format(response['message']))
|
||||
else:
|
||||
print("Short Link:\t{}".format(response['shorturl']))
|
||||
else:
|
||||
PBinCLIError("YOURLS: No status, statusCode or message fields in response! Received:\n{}".format(response))
|
||||
|
||||
|
||||
def _clckru(self, url):
|
||||
request = {'url': url}
|
||||
|
||||
try:
|
||||
result = self.session.post(
|
||||
url = "https://clck.ru/--",
|
||||
proxies = self.proxy,
|
||||
data = request)
|
||||
print("Short Link:\t{}".format(result.text))
|
||||
except Exception as ex:
|
||||
PBinCLIError("clck.ru: unexcepted behavior: {}".format(ex))
|
||||
|
||||
|
||||
def _tinyurl(self, url):
|
||||
request = {'url': url}
|
||||
|
||||
try:
|
||||
result = self.session.post(
|
||||
url = "https://tinyurl.com/api-create.php",
|
||||
proxies = self.proxy,
|
||||
data = request)
|
||||
print("Short Link:\t{}".format(result.text))
|
||||
except Exception as ex:
|
||||
PBinCLIError("TinyURL: unexcepted behavior: {}".format(ex))
|
||||
|
||||
|
||||
def _gd(self, url):
|
||||
request = {
|
||||
'format': 'json',
|
||||
'url': url,
|
||||
'logstats': 0 # we don't want use any statistics
|
||||
}
|
||||
headers = { 'User-Agent': self.useragent}
|
||||
|
||||
try:
|
||||
result = self.session.post(
|
||||
url = self.apiurl + "create.php",
|
||||
headers = headers,
|
||||
proxies = self.proxy,
|
||||
data = request)
|
||||
|
||||
response = result.json()
|
||||
|
||||
if 'shorturl' in response:
|
||||
print("Short Link:\t{}".format(response['shorturl']))
|
||||
else:
|
||||
PBinCLIError("{}: got error {} from API: {}".format(
|
||||
"is.gd" if self.api == 'isgd' else 'v.gd',
|
||||
response['errorcode'],
|
||||
response['errormessage']))
|
||||
|
||||
except Exception as ex:
|
||||
PBinCLIError("{}: unexcepted behavior: {}".format(
|
||||
"is.gd" if self.api == 'isgd' else 'v.gd',
|
||||
ex))
|
||||
|
||||
|
||||
def _cuttly(self, url):
|
||||
request = {
|
||||
'url': url,
|
||||
'domain': 0
|
||||
}
|
||||
|
||||
try:
|
||||
result = self.session.post(
|
||||
url = "https://cutt.ly/scripts/shortenUrl.php",
|
||||
proxies = self.proxy,
|
||||
data = request)
|
||||
print("Short Link:\t{}".format(result.text))
|
||||
except Exception as ex:
|
||||
PBinCLIError("cutt.ly: unexcepted behavior: {}".format(ex))
|
||||
|
||||
@@ -3,7 +3,7 @@ import os, sys, argparse
|
||||
|
||||
import pbincli.actions
|
||||
from pbincli.api import PrivateBin
|
||||
from pbincli.utils import PBinCLIException
|
||||
from pbincli.utils import PBinCLIException, validate_url
|
||||
|
||||
CONFIG_PATHS = [os.path.join(".", "pbincli.conf", ),
|
||||
os.path.join(os.getenv("HOME") or "~", ".config", "pbincli", "pbincli.conf") ]
|
||||
@@ -23,46 +23,75 @@ def main():
|
||||
subparsers = parser.add_subparsers(title="actions", help="List of commands")
|
||||
|
||||
# a send command
|
||||
send_parser = subparsers.add_parser("send", description="Send data to PrivateBin instance", usage="""
|
||||
%(prog)s --burn --discus --expire 1day --format plaintext \\
|
||||
--text "My file" --password mypass image.txt"""
|
||||
)
|
||||
send_parser.add_argument("-B", "--burn", default=False, action="store_true", help="burn sent paste after reading")
|
||||
send_parser.add_argument("-D", "--discus", default=False, action="store_true", help="open discussion of sent paste")
|
||||
send_parser = subparsers.add_parser("send", description="Send data to PrivateBin instance")
|
||||
send_parser.add_argument("-t", "--text", help="text in quotes. Ignored if used stdin. If not used, forcefully used stdin")
|
||||
send_parser.add_argument("-f", "--file", help="example: image.jpg or full path to file")
|
||||
send_parser.add_argument("-p", "--password", help="password for encrypting paste")
|
||||
send_parser.add_argument("-E", "--expire", default="1day", action="store",
|
||||
choices=["5min", "10min", "1hour", "1day", "1week", "1month", "1year", "never"], help="expiration of paste (default: 1day)")
|
||||
choices=["5min", "10min", "1hour", "1day", "1week", "1month", "1year", "never"], help="paste lifetime (default: 1day)")
|
||||
send_parser.add_argument("-B", "--burn", default=False, action="store_true", help="burn sent paste after reading")
|
||||
send_parser.add_argument("-D", "--discus", default=False, action="store_true", help="open discussion for sent paste")
|
||||
send_parser.add_argument("-F", "--format", default="plaintext", action="store",
|
||||
choices=["plaintext", "syntaxhighlighting", "markdown"], help="format of text (default: plaintext)")
|
||||
send_parser.add_argument("-t", "--text", help="comment in quotes. Ignored if used stdin")
|
||||
send_parser.add_argument("-p", "--password", help="password for encrypting paste")
|
||||
send_parser.add_argument("-q", "--notext", default=False, action="store_true", help="don't send text in paste")
|
||||
send_parser.add_argument("-c", "--compression", default="zlib", action="store",
|
||||
choices=["zlib", "none"], help="set compression for paste (default: zlib). Note: works only on v2 paste format")
|
||||
# URL shortener
|
||||
send_parser.add_argument("-S", "--short", default=False, action="store_true", help="use URL shortener")
|
||||
send_parser.add_argument("--short-api", default=argparse.SUPPRESS, action="store", choices=["tinyurl", "clckru", "isgd", "vgd", "cuttly", "yourls"], help="API used by shortener service")
|
||||
send_parser.add_argument("--short-url", default=argparse.SUPPRESS, help="URL of shortener service API")
|
||||
send_parser.add_argument("--short-user", default=argparse.SUPPRESS, help="Shortener username")
|
||||
send_parser.add_argument("--short-pass", default=argparse.SUPPRESS, help="Shortener password")
|
||||
send_parser.add_argument("--short-token", default=argparse.SUPPRESS, help="Shortener token")
|
||||
# Connection options
|
||||
send_parser.add_argument("-s", "--server", default=argparse.SUPPRESS, help="PrivateBin service URL (default: https://paste.i2pd.xyz/)")
|
||||
send_parser.add_argument("-x", "--proxy", default=argparse.SUPPRESS, help="Proxy server address (default: None)")
|
||||
send_parser.add_argument("--no-check-certificate", default=False, action="store_true", help="disable certificate validation")
|
||||
send_parser.add_argument("--no-insecure-warning", default=False, action="store_true", help="suppress InsecureRequestWarning (only with --no-check-certificate)")
|
||||
#
|
||||
send_parser.add_argument("-d", "--debug", default=False, action="store_true", help="enable debug")
|
||||
send_parser.add_argument("--dry", default=False, action="store_true", help="invoke dry run")
|
||||
send_parser.add_argument("-f", "--file", help="example: image.jpg or full path to file")
|
||||
|
||||
send_parser.add_argument("stdin", help="input paste text from stdin", nargs="?", type=argparse.FileType("r"), default=sys.stdin)
|
||||
send_parser.set_defaults(func=pbincli.actions.send)
|
||||
|
||||
get_parser = subparsers.add_parser("get", description="Get data from PrivateBin instance", usage="""
|
||||
%(prog)s pasteid#password"""
|
||||
)
|
||||
# a get command
|
||||
get_parser = subparsers.add_parser("get", description="Get data from PrivateBin instance")
|
||||
get_parser.add_argument("pasteinfo", help="example: aabb#cccddd")
|
||||
get_parser.add_argument("-d", "--debug", default=False, action="store_true", help="enable debug")
|
||||
get_parser.add_argument("-p", "--password", help="password for decrypting paste")
|
||||
get_parser.add_argument("--no-check-certificate", default=False, action="store_true", help="disable certificate validation")
|
||||
get_parser.add_argument("--no-insecure-warning", default=False, action="store_true", help="suppress InsecureRequestWarning (only with --no-check-certificate)")
|
||||
get_parser.add_argument("-d", "--debug", default=False, action="store_true", help="enable debug")
|
||||
get_parser.set_defaults(func=pbincli.actions.get)
|
||||
|
||||
delete_parser = subparsers.add_parser("delete", description="Delete paste from PrivateBin instance using token", usage="""
|
||||
%(prog)s --paste aabb --token aabbcc"""
|
||||
)
|
||||
# a delete command
|
||||
delete_parser = subparsers.add_parser("delete", description="Delete paste from PrivateBin instance using token")
|
||||
delete_parser.add_argument("-p", "--paste", required=True, help="paste id")
|
||||
delete_parser.add_argument("-t", "--token", required=True, help="delete token")
|
||||
delete_parser.add_argument("-t", "--token", required=True, help="paste deletion token")
|
||||
delete_parser.add_argument("--no-check-certificate", default=False, action="store_true", help="disable certificate validation")
|
||||
delete_parser.add_argument("--no-insecure-warning", default=False, action="store_true", help="suppress InsecureRequestWarning (only with --no-check-certificate)")
|
||||
delete_parser.add_argument("-d", "--debug", default=False, action="store_true", help="enable debug")
|
||||
delete_parser.set_defaults(func=pbincli.actions.delete)
|
||||
|
||||
# parse arguments
|
||||
args = parser.parse_args()
|
||||
|
||||
CONFIG = {"server": "https://paste.i2pd.xyz/",
|
||||
"proxy": None}
|
||||
CONFIG = {
|
||||
'server': 'https://paste.i2pd.xyz/',
|
||||
'proxy': None,
|
||||
'short_api': None,
|
||||
'short_url': None,
|
||||
'short_user': None,
|
||||
'short_pass': None,
|
||||
'short_token': None,
|
||||
'no_check_certificate': False,
|
||||
'no_insecure_warning': False
|
||||
}
|
||||
|
||||
# Configuration preference order:
|
||||
# 1. Command line switches
|
||||
# 2. Environment variables
|
||||
# 3. Configuration file
|
||||
# 4. Default values below
|
||||
|
||||
for p in CONFIG_PATHS:
|
||||
if os.path.exists(p):
|
||||
@@ -72,15 +101,21 @@ def main():
|
||||
for key in CONFIG.keys():
|
||||
var = "PRIVATEBIN_{}".format(key.upper())
|
||||
if var in os.environ: CONFIG[key] = os.getenv(var)
|
||||
# values from command line switches are preferred
|
||||
args_var = vars(args)
|
||||
if key in args_var:
|
||||
CONFIG[key] = args_var[key]
|
||||
|
||||
api_client = PrivateBin(CONFIG["server"], proxy=CONFIG["proxy"])
|
||||
# Re-validate PrivateBin instance URL
|
||||
CONFIG['server'] = validate_url(CONFIG['server'])
|
||||
|
||||
api_client = PrivateBin(CONFIG)
|
||||
|
||||
if hasattr(args, "func"):
|
||||
try:
|
||||
args.func(args, api_client)
|
||||
args.func(args, api_client, settings=CONFIG)
|
||||
except PBinCLIException as pe:
|
||||
print("PBinCLI error: {}".format(pe))
|
||||
sys.exit(1)
|
||||
raise PBinCLIException("error: {}".format(pe))
|
||||
else:
|
||||
parser.print_help()
|
||||
|
||||
|
||||
290
pbincli/format.py
Normal file
290
pbincli/format.py
Normal file
@@ -0,0 +1,290 @@
|
||||
from Crypto.Random import get_random_bytes
|
||||
from Crypto.Cipher import AES
|
||||
from base64 import b64encode, b64decode
|
||||
from pbincli.utils import PBinCLIError
|
||||
import zlib
|
||||
|
||||
CIPHER_ITERATION_COUNT = 100000
|
||||
CIPHER_SALT_BYTES = 8
|
||||
CIPHER_BLOCK_BITS = 256
|
||||
CIPHER_BLOCK_BYTES = int(CIPHER_BLOCK_BITS/8)
|
||||
CIPHER_TAG_BITS = int(CIPHER_BLOCK_BITS/2)
|
||||
CIPHER_TAG_BYTES = int(CIPHER_TAG_BITS/8)
|
||||
|
||||
class Paste:
|
||||
def __init__(self, debug=False):
|
||||
self._version = 2
|
||||
self._compression = 'zlib'
|
||||
self._data = ''
|
||||
self._text = ''
|
||||
self._attachment = ''
|
||||
self._attachment_name = ''
|
||||
self._key = get_random_bytes(CIPHER_BLOCK_BYTES)
|
||||
self._password = ''
|
||||
self._debug = debug
|
||||
|
||||
|
||||
def setVersion(self, version):
|
||||
if self._debug: print("Set paste version to {}".format(version))
|
||||
self._version = version
|
||||
|
||||
|
||||
def setPassword(self, password):
|
||||
self._password = password
|
||||
|
||||
|
||||
def setText(self, text):
|
||||
self._text = text
|
||||
|
||||
|
||||
def setAttachment(self, path):
|
||||
from pbincli.utils import check_readable, path_leaf
|
||||
from mimetypes import guess_type
|
||||
|
||||
check_readable(path)
|
||||
with open(path, 'rb') as f:
|
||||
contents = f.read()
|
||||
f.close()
|
||||
mime = guess_type(path, strict=False)[0]
|
||||
|
||||
# MIME fallback
|
||||
if not mime: mime = 'application/octet-stream'
|
||||
|
||||
if self._debug: print("Filename:\t{}\nMIME-type:\t{}".format(path_leaf(path), mime))
|
||||
|
||||
self._attachment = 'data:' + mime + ';base64,' + b64encode(contents).decode()
|
||||
self._attachment_name = path_leaf(path)
|
||||
|
||||
def setCompression(self, comp):
|
||||
self._compression = comp
|
||||
|
||||
|
||||
def getText(self):
|
||||
return self._text
|
||||
|
||||
|
||||
def getAttachment(self):
|
||||
return [b64decode(self._attachment.split(',', 1)[1]), self._attachment_name] \
|
||||
if self._attachment \
|
||||
else [False,False]
|
||||
|
||||
|
||||
def getJSON(self):
|
||||
if self._version == 2:
|
||||
from pbincli.utils import json_encode
|
||||
return json_encode(self._data).decode()
|
||||
else:
|
||||
return self._data
|
||||
|
||||
|
||||
def loadJSON(self, data):
|
||||
self._data = data
|
||||
|
||||
|
||||
def getHash(self):
|
||||
if self._version == 2:
|
||||
from base58 import b58encode
|
||||
return b58encode(self._key).decode()
|
||||
else:
|
||||
return b64encode(self._key).decode()
|
||||
|
||||
|
||||
def setHash(self, passphrase):
|
||||
if self._version == 2:
|
||||
from base58 import b58decode
|
||||
self._key = b58decode(passphrase)
|
||||
else:
|
||||
self._key = b64decode(passphrase)
|
||||
|
||||
|
||||
def __deriveKey(self, salt):
|
||||
from Crypto.Protocol.KDF import PBKDF2
|
||||
from Crypto.Hash import HMAC, SHA256
|
||||
|
||||
# Key derivation, using PBKDF2 and SHA256 HMAC
|
||||
return PBKDF2(
|
||||
self._key + self._password.encode(),
|
||||
salt,
|
||||
dkLen = CIPHER_BLOCK_BYTES,
|
||||
count = CIPHER_ITERATION_COUNT,
|
||||
prf = lambda password, salt: HMAC.new(
|
||||
password,
|
||||
salt,
|
||||
SHA256
|
||||
).digest())
|
||||
|
||||
|
||||
@classmethod
|
||||
def __initializeCipher(self, key, iv, adata):
|
||||
from pbincli.utils import json_encode
|
||||
|
||||
cipher = AES.new(key, AES.MODE_GCM, nonce=iv, mac_len=CIPHER_TAG_BYTES)
|
||||
cipher.update(json_encode(adata))
|
||||
return cipher
|
||||
|
||||
|
||||
def __preparePassKey(self):
|
||||
from hashlib import sha256
|
||||
|
||||
if self._password:
|
||||
digest = sha256(self._password.encode("UTF-8")).hexdigest()
|
||||
return b64encode(self._key) + digest.encode("UTF-8")
|
||||
else:
|
||||
return b64encode(self._key)
|
||||
|
||||
|
||||
def __decompress(self, s):
|
||||
if self._version == 2 and self._compression == 'zlib':
|
||||
# decompress data
|
||||
return zlib.decompress(s, -zlib.MAX_WBITS)
|
||||
elif self._version == 2 and self._compression == 'none':
|
||||
# nothing to do, just return original data
|
||||
return s
|
||||
elif self._version == 1:
|
||||
return zlib.decompress(bytearray(map(lambda c:ord(c)&255, b64decode(s.encode('utf-8')).decode('utf-8'))), -zlib.MAX_WBITS)
|
||||
else:
|
||||
PBinCLIError('Unknown compression type provided in paste!')
|
||||
|
||||
|
||||
def __compress(self, s):
|
||||
if self._version == 2 and self._compression == 'zlib':
|
||||
# using compressobj as compress doesn't let us specify wbits
|
||||
# needed to get the raw stream without headers
|
||||
co = zlib.compressobj(wbits=-zlib.MAX_WBITS)
|
||||
return co.compress(s) + co.flush()
|
||||
elif self._version == 2 and self._compression == 'none':
|
||||
# nothing to do, just return original data
|
||||
return s
|
||||
elif self._version == 1:
|
||||
co = zlib.compressobj(wbits=-zlib.MAX_WBITS)
|
||||
b = co.compress(s) + co.flush()
|
||||
return b64encode(''.join(map(chr, b)).encode('utf-8'))
|
||||
else:
|
||||
PBinCLIError('Unknown compression type provided!')
|
||||
|
||||
|
||||
def decrypt(self):
|
||||
# that is wrapper which running needed function regrading to paste version
|
||||
if self._version == 2: self._decryptV2()
|
||||
else: self._decryptV1()
|
||||
|
||||
|
||||
def _decryptV2(self):
|
||||
from json import loads as json_decode
|
||||
iv = b64decode(self._data['adata'][0][0])
|
||||
salt = b64decode(self._data['adata'][0][1])
|
||||
key = self.__deriveKey(salt)
|
||||
|
||||
# Get compression type from received paste
|
||||
self._compression = self._data['adata'][0][7]
|
||||
|
||||
cipher = self.__initializeCipher(key, iv, self._data['adata'])
|
||||
# Cut the cipher text into message and tag
|
||||
cipher_text_tag = b64decode(self._data['ct'])
|
||||
cipher_text = cipher_text_tag[:-CIPHER_TAG_BYTES]
|
||||
cipher_tag = cipher_text_tag[-CIPHER_TAG_BYTES:]
|
||||
cipher_message = json_decode(self.__decompress(cipher.decrypt_and_verify(cipher_text, cipher_tag)).decode())
|
||||
|
||||
self._text = cipher_message['paste'].encode()
|
||||
|
||||
if 'attachment' in cipher_message and 'attachment_name' in cipher_message:
|
||||
self._attachment = cipher_message['attachment']
|
||||
self._attachment_name = cipher_message['attachment_name']
|
||||
|
||||
|
||||
def _decryptV1(self):
|
||||
from sjcl import SJCL
|
||||
from json import loads as json_decode
|
||||
|
||||
password = self.__preparePassKey()
|
||||
cipher_text = json_decode(self._data['data'])
|
||||
if self._debug: print("Text:\t{}\n".format(cipher_text))
|
||||
|
||||
text = SJCL().decrypt(cipher_text, password)
|
||||
|
||||
if len(text):
|
||||
if self._debug: print("Decoded Text:\t{}\n".format(text))
|
||||
self._text = self.__decompress(text.decode())
|
||||
|
||||
if 'attachment' in self._data and 'attachmentname' in self._data:
|
||||
cipherfile = json_decode(self._data['attachment'])
|
||||
cipherfilename = json_decode(self._data['attachmentname'])
|
||||
|
||||
if self._debug: print("Name:\t{}\nData:\t{}".format(cipherfilename, cipherfile))
|
||||
|
||||
attachment = SJCL().decrypt(cipherfile, password)
|
||||
attachmentname = SJCL().decrypt(cipherfilename, password)
|
||||
|
||||
self._attachment = self.__decompress(attachment.decode('utf-8')).decode('utf-8')
|
||||
self._attachment_name = self.__decompress(attachmentname.decode('utf-8')).decode('utf-8')
|
||||
|
||||
|
||||
def encrypt(self, formatter, burnafterreading, discussion, expiration):
|
||||
# that is wrapper which running needed function regrading to paste version
|
||||
self._formatter = formatter
|
||||
self._burnafterreading = burnafterreading
|
||||
self._discussion = discussion
|
||||
self._expiration = expiration
|
||||
|
||||
if self._version == 2: self._encryptV2()
|
||||
else: self._encryptV1()
|
||||
|
||||
|
||||
def _encryptV2(self):
|
||||
from pbincli.utils import json_encode
|
||||
|
||||
iv = get_random_bytes(CIPHER_TAG_BYTES)
|
||||
salt = get_random_bytes(CIPHER_SALT_BYTES)
|
||||
key = self.__deriveKey(salt)
|
||||
|
||||
# prepare encryption authenticated data and message
|
||||
adata = [
|
||||
[
|
||||
b64encode(iv).decode(),
|
||||
b64encode(salt).decode(),
|
||||
CIPHER_ITERATION_COUNT,
|
||||
CIPHER_BLOCK_BITS,
|
||||
CIPHER_TAG_BITS,
|
||||
'aes',
|
||||
'gcm',
|
||||
self._compression
|
||||
],
|
||||
self._formatter,
|
||||
int(self._discussion),
|
||||
int(self._burnafterreading)
|
||||
]
|
||||
cipher_message = {'paste':self._text}
|
||||
if self._attachment:
|
||||
cipher_message['attachment'] = self._attachment
|
||||
cipher_message['attachment_name'] = self._attachment_name
|
||||
|
||||
cipher = self.__initializeCipher(key, iv, adata)
|
||||
ciphertext, tag = cipher.encrypt_and_digest(self.__compress(json_encode(cipher_message)))
|
||||
|
||||
self._data = {'v':2,'adata':adata,'ct':b64encode(ciphertext + tag).decode(),'meta':{'expire':self._expiration}}
|
||||
|
||||
|
||||
def _encryptV1(self):
|
||||
from sjcl import SJCL
|
||||
from pbincli.utils import json_encode
|
||||
|
||||
self._data = {'expire':self._expiration,'formatter':self._formatter,'burnafterreading':int(self._burnafterreading),'opendiscussion':int(self._discussion)}
|
||||
|
||||
password = self.__preparePassKey()
|
||||
if self._debug: print("Password:\t{}".format(password))
|
||||
|
||||
# Encrypting text
|
||||
cipher = SJCL().encrypt(self.__compress(self._text.encode('utf-8')), password, mode='gcm')
|
||||
for k in ['salt', 'iv', 'ct']: cipher[k] = cipher[k].decode()
|
||||
|
||||
self._data['data'] = json_encode(cipher)
|
||||
|
||||
if self._attachment:
|
||||
cipherfile = SJCL().encrypt(self.__compress(self._attachment.encode('utf-8')), password, mode='gcm')
|
||||
for k in ['salt', 'iv', 'ct']: cipherfile[k] = cipherfile[k].decode()
|
||||
|
||||
cipherfilename = SJCL().encrypt(self.__compress(self._attachment_name.encode('utf-8')), password, mode='gcm')
|
||||
for k in ['salt', 'iv', 'ct']: cipherfilename[k] = cipherfilename[k].decode()
|
||||
|
||||
self._data['attachment'] = json_encode(cipherfile)
|
||||
self._data['attachmentname'] = json_encode(cipherfilename)
|
||||
@@ -1,50 +1,36 @@
|
||||
import json, os
|
||||
|
||||
import json, ntpath, os, sys
|
||||
|
||||
class PBinCLIException(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def PBinCLIError(message):
|
||||
print("PBinCLI Error: {}".format(message), file=sys.stderr)
|
||||
exit(1)
|
||||
|
||||
|
||||
def path_leaf(path):
|
||||
head, tail = ntpath.split(path)
|
||||
return tail or ntpath.basename(head)
|
||||
|
||||
|
||||
def check_readable(f):
|
||||
# Checks if path exists and readable
|
||||
if not os.path.exists(f) or not os.access(f, os.R_OK):
|
||||
raise PBinCLIException("Error accessing path: {}".format(f))
|
||||
PBinCLIError("Error accessing path: {}".format(f))
|
||||
|
||||
|
||||
def check_writable(f):
|
||||
# Checks if path is writable
|
||||
if not os.access(os.path.dirname(f) or ".", os.W_OK):
|
||||
raise PBinCLIException("Path is not writable: {}".format(f))
|
||||
PBinCLIError("Path is not writable: {}".format(f))
|
||||
|
||||
|
||||
# http://stackoverflow.com/a/33571117
|
||||
def json_load_byteified(file_handle):
|
||||
return _byteify(
|
||||
json.load(file_handle, object_hook=_byteify),
|
||||
ignore_dicts=True
|
||||
)
|
||||
def json_encode(s):
|
||||
return json.dumps(s, separators=(',',':')).encode()
|
||||
|
||||
|
||||
def json_loads_byteified(json_text):
|
||||
return _byteify(
|
||||
json.loads(json_text, object_hook=_byteify),
|
||||
ignore_dicts=True
|
||||
)
|
||||
|
||||
|
||||
def _byteify(data, ignore_dicts = False):
|
||||
# if this is a unicode string, return its string representation
|
||||
if isinstance(data, str):
|
||||
return data.encode('utf-8')
|
||||
# if this is a list of values, return list of byteified values
|
||||
if isinstance(data, list):
|
||||
return [ _byteify(item, ignore_dicts=True) for item in data ]
|
||||
# if this is a dictionary, return dictionary of byteified keys and values
|
||||
# but only if we haven't already byteified it
|
||||
if isinstance(data, dict) and not ignore_dicts:
|
||||
return {
|
||||
_byteify(key, ignore_dicts=True): _byteify(value, ignore_dicts=True)
|
||||
for key, value in data.iteritems()
|
||||
}
|
||||
# if it's anything else, return it in its original form
|
||||
return data
|
||||
def validate_url(s):
|
||||
if not s.endswith('/'):
|
||||
s = s + "/"
|
||||
return s
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
pycryptodome
|
||||
sjcl
|
||||
base58
|
||||
requests
|
||||
|
||||
19
setup.py
19
setup.py
@@ -1,8 +1,9 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
from setuptools import setup
|
||||
from pbincli.__init__ import __version__ as pbincli_version
|
||||
|
||||
with open("README.md") as readme:
|
||||
with open("README.rst") as readme:
|
||||
long_description = readme.read()
|
||||
|
||||
with open("requirements.txt") as f:
|
||||
@@ -10,19 +11,23 @@ with open("requirements.txt") as f:
|
||||
|
||||
setup(
|
||||
name='PBinCLI',
|
||||
version='0.1',
|
||||
version=pbincli_version,
|
||||
description='PrivateBin client for command line',
|
||||
long_description=long_description,
|
||||
long_description_content_type='text/x-rst',
|
||||
author='R4SAS',
|
||||
author_email='r4sas@i2pmail.org',
|
||||
url='https://github.com/r4sas/PBinCLI',
|
||||
keywords='privatebin',
|
||||
license='DWTFYWWI',
|
||||
keywords='privatebin cryptography security',
|
||||
license='MIT',
|
||||
classifiers=[
|
||||
'Development Status :: 4 - Beta',
|
||||
'License :: Other/Proprietary License',
|
||||
'Programming Language :: Python :: 3',
|
||||
'Development Status :: 5 - Production/Stable',
|
||||
'Environment :: Console',
|
||||
'Intended Audience :: End Users/Desktop',
|
||||
'License :: OSI Approved :: MIT License',
|
||||
'Programming Language :: Python :: 3',
|
||||
'Topic :: Security :: Cryptography',
|
||||
'Topic :: Utilities',
|
||||
],
|
||||
packages=['pbincli'],
|
||||
install_requires=install_requires,
|
||||
|
||||
Reference in New Issue
Block a user