49 Commits
0.1 ... 0.3.0

Author SHA1 Message Date
r4sas
682b47fbd3 0.3.0
New:
* Added validation of PrivateBin instance URL - #18 (it must contain trailing slash because POST is used)
* URL shortener support with various supported services - #19
* Shortener configuration, certificate validation and insecure warning settings can be configured in config file or via env

Changed:
* Restructured some parts of code by splitting big code chunks in funtions (encrypt/decrypt)
* Rework error messaging repeatable code (moved in utils)
* Reduce code duplication (requests session configuring)

Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-20 10:50:06 +00:00
r4sas
9d82c727b6 [shortener] move service-depend init code to functions
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-20 06:12:10 +00:00
r4sas
c425d86ed6 [shortener] remove not completed bitly support code
Anyone can request to add support later, so for now I removing that code due to lack of use.

Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-20 06:00:08 +00:00
r4sas
3ed06686ab Codacy: fix W0106
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-19 00:10:39 +00:00
r4sas
7e4fb0a8c4 split encrypt and decrypt code in separated functions by paste version
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 23:48:46 +00:00
r4sas
ebfe0c48a0 [shortener] separate services related code in functions
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 23:22:56 +00:00
r4sas
45d854e590 [shortener] add is.gd, v.gd and cutt.ly services support
Start realisation of bitly support.

Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 21:09:16 +00:00
r4sas
18d79c8e04 add .gitattributes
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 19:07:15 +00:00
r4sas
92c38344e3 add tinyurl support
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 19:01:41 +00:00
r4sas
b596f42b7e fix response handler (closes #19)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 13:46:30 +00:00
r4sas
86c5051fcf move 'requests' configuration code in separate function
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 12:02:41 +00:00
r4sas
f838f8ee94 fix codacy issues
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 11:43:44 +00:00
r4sas
fb7a93732d rewrite yourls response handler (#19)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 11:27:15 +00:00
r4sas
181763070c remove unused import, replace comment definition
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 10:22:58 +00:00
r4sas
c3a491ac46 correct response store for yourls (#19)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-18 10:14:44 +00:00
r4sas
641c55a6a2 fix dict merging (#19)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-17 13:22:28 +00:00
r4sas
432675f2e6 url shortener support (#19)
Currently tested only with clck.ru service. YOURLS test needed.

Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-17 10:54:58 +00:00
r4sas
635c87dabd [wip] url shortener support (#19)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-17 08:55:59 +00:00
r4sas
6b6c33e545 [wip] url shortener support (#19)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-17 05:22:57 +00:00
r4sas
7c5ba2fdbe add URL validation for trailing slash (closes #18)
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-09-09 14:23:30 +00:00
r4sas
692335ee62 0.2.1
Signed-off-by: r4sas <r4sas@i2pmail.org>
2019-08-16 21:24:52 +00:00
R4SAS
c63256c628 Merge pull request #16 from firecat53/patch-1
Fix swapped "open discussion" and "burn after reading" options order
2019-08-17 00:12:49 +03:00
Scott Hansen
66659da66d Update format.py
The --burn and --discuss flags were switched. This change fixes that.
2019-08-16 12:59:36 -07:00
r4sas
32d60b5321 0.2 2019-07-24 11:22:37 +00:00
r4sas
9e9f480075 add description content type 2019-07-16 20:52:15 +00:00
r4sas
7fc2a1a625 add cert validation ignoring switches (closes #15) 2019-07-16 20:32:37 +00:00
r4sas
8aea956e77 Update function in map iterator
Fixes #14.
Thanks to @mqus for solution.
Taken from https://github.com/PrivateBin/PrivateBin/pull/193#issuecomment-480262590
2019-06-24 01:18:27 +00:00
r4sas
ac58fff9ce fix classifiers 2019-06-22 21:13:58 +00:00
r4sas
7a59d1acd1 0.2-beta1 2019-06-21 16:15:37 +00:00
r4sas
5c6b9611d8 fix errors 2019-06-21 14:35:45 +00:00
r4sas
dc034b1d55 update rst readme, remove unused import 2019-06-21 13:32:46 +00:00
r4sas
5d988e01fc add codacy badge 2019-06-21 13:15:13 +00:00
r4sas
85cc1454ea fixes 2019-06-21 13:12:17 +00:00
r4sas
f5ef4bbc03 fix v1 decrypt debug message, update readme 2019-06-21 12:43:59 +00:00
r4sas
8d7a9235b8 add code comments 2019-06-21 11:59:23 +00:00
r4sas
70f386193a little refactoring 2019-06-21 11:49:20 +00:00
r4sas
d37e573d9e little refactoring 2019-06-19 10:58:41 +00:00
r4sas
e850b5495a load zlib globally in format 2019-06-19 10:33:11 +00:00
r4sas
9390edeb79 reorder argparse, clarify option descriptions 2019-06-07 11:11:53 +00:00
r4sas
d11beb10af Make compression optional
ref: https://github.com/PrivateBin/PrivateBin/issues/38
Compression is optional only in v2 paste format.
Currently available types: 'zlib' and 'none'.
To set compression use `-c "none"` or `--compression "none".

When receiving paste, compression detected from 'adata' field
2019-06-07 09:16:58 +00:00
r4sas
4c124a33c0 [v2] encode json for v2, move de/compress to Paste class (#13) 2019-06-03 07:18:07 +00:00
r4sas
5b38c532a2 [wip][v2] json bytes2str in decode (#13) 2019-06-02 14:27:53 +00:00
r4sas
f7fae450a0 [wip] v2 support code (#13) 2019-06-02 14:04:38 +00:00
r4sas
1ff6e721c7 move functions to utils, import functions only where they needed, remove bytified json load 2019-06-01 15:09:57 +00:00
r4sas
487def2b45 add notext option, fix paste without text handling
Now text is taken if no `--notext (-q)` argument is passed.

If that flag used, paste will be created with empty text,
but file for send must be specified

If `notext` arg is not used, text will be taken in next order:
* from `--text (-t)` argument
* from stdin (forcibly)

Here's no ability create paste without any text.
2019-06-01 14:26:17 +00:00
r4sas
c27606d442 update gitignore, add tag button in readme 2019-05-31 14:42:22 +00:00
r4sas
2381893b72 use rst readme as long description 2019-05-31 12:52:04 +00:00
r4sas
03b4fcc2ea update information 2019-05-31 12:44:14 +00:00
r4sas
587dec6b6a let's use normal license 2019-05-31 12:04:59 +00:00
13 changed files with 993 additions and 285 deletions

3
.gitattributes vendored Normal file
View File

@@ -0,0 +1,3 @@
.gitattributes export-ignore
.gitignore export-ignore
README.md export-ignore

133
.gitignore vendored
View File

@@ -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/ 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
View File

@@ -1,17 +1,7 @@
DWTFYWWI LICENSE Copyright 2017 © R4SAS <r4sas@i2pmail.org>
Version 1, January 2006
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 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.
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.

View File

@@ -1,55 +1,67 @@
[![GitHub license](https://img.shields.io/github/license/r4sas/PBinCLI.svg)](https://github.com/r4sas/PBinCLI/blob/master/LICENSE)
[![GitHub tag](https://img.shields.io/github/tag/r4sas/PBinCLI.svg)](https://github.com/r4sas/PBinCLI/tags/)
[![Codacy Badge](https://api.codacy.com/project/badge/Grade/4f24f43356a84621bbd9078c4b3f1b70)](https://www.codacy.com/app/r4sas/PBinCLI?utm_source=github.com&amp;utm_medium=referral&amp;utm_content=r4sas/PBinCLI&amp;utm_campaign=Badge_Grade)
PBinCLI 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 Installing
----- -----
```bash ```bash
$ virtualenv --python=python3 venv virtualenv --python=python3 venv
$ . venv/bin/activate . venv/bin/activate
$ pip install pbincli pip install pbincli
``` ```
Usage 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. You can create config file with variables `server` and `proxy` in `~/.config/pbincli/pbincli.conf` to use different settings.
Example contents: Example contents:
``` ```ini
server=https://paste.i2pd.xyz/ server=https://paste.i2pd.xyz/
proxy=http://127.0.0.1:3128 proxy=http://127.0.0.1:3128
``` ```
Run inside `venv` command: Run inside `venv` command:
$ pbincli send --text "Hello!" ```bash
pbincli send --text "Hello!"
```
Or use stdin input to read text for paste: Or use stdin input to read text for paste:
$ pbincli send - <<EOF ```bash
Hello! This is test paste! pbincli send - <<EOF
EOF Hello! This is test paste!
EOF
```
It will send string `Hello! This is test paste!` to PrivateBin. It will send string `Hello! This is test paste!` to PrivateBin.
To send file use `--file` or `-f` with filename. Example: 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. To retrieve paste from server, use `get` command with paste info.
It must be formated like `pasteID#passphrase`. Example: 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 More info you can find by typing
$ pbincli [-h] {send, get, delete} ```bash
pbincli [-h] {send, get, delete}
```
TODO TODO
---- ----
@@ -57,5 +69,5 @@ Write a more complete usage documentation.
License License
------- -------
This project is licensed under the DWTFYWWI license, which can be found in the file This project is licensed under the MIT license, which can be found in the file
[LICENSE](LICENSE) in the root of the project source code. [LICENSE](https://github.com/r4sas/PBinCLI/blob/master/LICENSE) in the root of the project source code.

91
README.rst Normal file
View 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&amp;utm_medium=referral&amp;utm_content=r4sas/PBinCLI&amp;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.

View File

@@ -2,6 +2,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
__author__ = "R4SAS <r4sas@i2pmail.org>" __author__ = "R4SAS <r4sas@i2pmail.org>"
__version__ = "0.1" __version__ = "0.3.0"
__copyright__ = "Copyright (c) R4SAS" __copyright__ = "Copyright (c) R4SAS"
__license__ = "DWTFYWWI" __license__ = "MIT"

View File

@@ -1,219 +1,162 @@
import json, hashlib, ntpath, os, sys, zlib from pbincli.format import Paste
import pbincli.actions from pbincli.utils import PBinCLIError
from sjcl import SJCL
from base64 import b64encode, b64decode def send(args, api_client, settings=None):
from mimetypes import guess_type from pbincli.api import Shortener
from pbincli.utils import PBinCLIException, check_readable, check_writable, json_load_byteified if args.short:
shortener = Shortener(settings)
if not args.notext:
def path_leaf(path): if args.text:
head, tail = ntpath.split(path) text = args.text
return tail or ntpath.basename(head) elif args.stdin:
text = args.stdin.read()
elif not args.file:
def decompress(s): PBinCLIError("Nothing to send!")
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!"
else: else:
print("Nothing to send!") text = ""
sys.exit(1)
# Formatting request paste = Paste(args.debug)
request = {'expire':args.expire,'formatter':args.format,'burnafterreading':int(args.burn),'opendiscussion':int(args.discus)}
passphrase = b64encode(os.urandom(32)) # get from server supported paste format version and update object
if args.debug: print("Passphrase:\t{}".format(passphrase)) 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 we set PASSWORD variable
if args.password: if args.password:
digest = hashlib.sha256(args.password.encode("UTF-8")).hexdigest() paste.setPassword(args.password)
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(' ','')
# If we set FILE variable # If we set FILE variable
if args.file: if args.file:
check_readable(args.file) paste.setAttachment(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]))
file = "data:" + mime[0] + ";base64," + b64encode(contents).decode() paste.encrypt(
filename = path_leaf(args.file) formatter = args.format,
burnafterreading = args.burn,
discussion = args.discus,
expiration = args.expire)
cipherfile = SJCL().encrypt(compress(file.encode('utf-8')), password, mode='gcm') request = paste.getJSON()
# 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['attachment'] = json.dumps(cipherfile, ensure_ascii=False).replace(' ','') if args.debug:
request['attachmentname'] = json.dumps(cipherfilename, ensure_ascii=False).replace(' ','') print("Passphrase:\t{}".format(paste.getHash()))
print("Request:\t{}".format(request))
if args.debug: print("Request:\t{}".format(request))
# If we use dry option, exit now # If we use dry option, exit now
if args.dry: sys.exit(0) if args.dry: exit(0)
result = api_client.post(request) result = api_client.post(request)
if args.debug: print("Response:\t{}\n".format(result)) 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: try:
result = json.loads(result) pasteid, passphrase = args.pasteinfo.split("#")
except ValueError as e: except ValueError:
print("PBinCLI Error: {}".format(e)) PBinCLIError("Provided info hasn't contain valid PasteID#Passphrase string")
sys.exit(1)
if 'status' in result and not result['status']: if not (pasteid and passphrase):
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())) PBinCLIError("Incorrect request")
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 args.debug: print("PasteID:\t{}\nPassphrase:\t{}".format(pasteid, passphrase))
def get(args, api_client): paste = Paste(args.debug)
pasteid, passphrase = args.pasteinfo.split("#")
if pasteid and passphrase: if args.password:
if args.debug: print("PasteID:\t{}\nPassphrase:\t{}".format(pasteid, passphrase)) paste.setPassword(args.password)
if args.debug: print("Password:\t{}".format(args.password))
if args.password: result = api_client.get(pasteid)
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)
if args.debug: print("Response:\t{}\n".format(result)) if args.debug: print("Response:\t{}\n".format(result))
try: # Paste was received. Checking received status code
result = json.loads(result) if not result['status']: # return code is zero
except ValueError as e: print("Paste received!")
print("PBinCLI Error: {}".format(e))
sys.exit(1)
if 'status' in result and not result['status']: version = result['v'] if 'v' in result else 1
print("Paste received! Text inside:") paste.setVersion(version)
data = json.loads(result['data'])
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) paste.setHash(passphrase)
print("{}\n".format(decompress(text.decode()))) paste.loadJSON(result)
paste.decrypt()
check_writable("paste.txt") text = paste.getText()
with open("paste.txt", "wb") as f:
f.write(decompress(text.decode()))
f.close
if 'attachment' in result and 'attachmentname' in result: if args.debug: print("Decoded text size: {}\n".format(len(text)))
print("Found file, attached to paste. Decoding it and saving")
cipherfile = json.loads(result['attachment']) if len(text):
cipherfilename = json.loads(result['attachmentname']) if args.debug: print("{}\n".format(text.decode()))
filename = "paste-" + pasteid + ".txt"
if args.debug: print("Name:\t{}\nData:\t{}".format(cipherfilename, cipherfile)) print("Found text in paste. Saving it to {}".format(filename))
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))
check_writable(filename) check_writable(filename)
with open(filename, "wb") as f: with open(filename, "wb") as f:
f.write(file) f.write(text)
f.close 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...") 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)) elif result['status']: # return code is other then zero
PBinCLIError("Something went wrong...\nError:\t\t{}".format(result['message']))
try: else: # or here no status field in response or it is empty
result = json.loads(result) PBinCLIError("Something went wrong...\nError: Empty response.")
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)
def delete(args, api_client): def delete(args, api_client, settings=None):
from pbincli.utils import json_encode
pasteid = args.paste pasteid = args.paste
token = args.token token = args.token
if args.debug: print("PasteID:\t{}\nToken:\t\t{}".format(pasteid, token)) if args.debug: print("PasteID:\t{}\nToken:\t\t{}".format(pasteid, token))
result = api_client.delete(pasteid, token) api_client.delete(json_encode({'pasteid':pasteid,'deletetoken':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)

View File

@@ -1,27 +1,248 @@
import requests 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: class PrivateBin:
def __init__(self, server, proxy=None): def __init__(self, settings=None):
self.server = server self.server = settings['server']
self.headers = {'X-Requested-With': 'JSONHttpRequest'} 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): def post(self, request):
r = requests.post(url = self.server, headers = self.headers, proxies = self.proxy, data = request) result = self.session.post(
return r.text 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): def get(self, request):
url = self.server + "?" + request return self.session.get(
r = requests.get(url = url, headers = self.headers, proxies = self.proxy) url = self.server + "?" + request,
return r.text headers = self.headers,
proxies = self.proxy).json()
def delete(self, pasteid, token): def delete(self, request):
request = {'pasteid':pasteid,'deletetoken':token} # using try as workaround for versions < 1.3 due to we cant detect
r = requests.post(url = self.server, headers = self.headers, proxies = self.proxy, data = request) # if server used version 1.2, where auto-deletion is added
return r.text 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))

View File

@@ -3,7 +3,7 @@ import os, sys, argparse
import pbincli.actions import pbincli.actions
from pbincli.api import PrivateBin from pbincli.api import PrivateBin
from pbincli.utils import PBinCLIException from pbincli.utils import PBinCLIException, validate_url
CONFIG_PATHS = [os.path.join(".", "pbincli.conf", ), CONFIG_PATHS = [os.path.join(".", "pbincli.conf", ),
os.path.join(os.getenv("HOME") or "~", ".config", "pbincli", "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") subparsers = parser.add_subparsers(title="actions", help="List of commands")
# a send command # a send command
send_parser = subparsers.add_parser("send", description="Send data to PrivateBin instance", usage=""" send_parser = subparsers.add_parser("send", description="Send data to PrivateBin instance")
%(prog)s --burn --discus --expire 1day --format plaintext \\ send_parser.add_argument("-t", "--text", help="text in quotes. Ignored if used stdin. If not used, forcefully used stdin")
--text "My file" --password mypass image.txt""" 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("-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.add_argument("-E", "--expire", default="1day", action="store", 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", send_parser.add_argument("-F", "--format", default="plaintext", action="store",
choices=["plaintext", "syntaxhighlighting", "markdown"], help="format of text (default: plaintext)") 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("-q", "--notext", default=False, action="store_true", help="don't send text in paste")
send_parser.add_argument("-p", "--password", help="password for encrypting 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("-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("--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.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) send_parser.set_defaults(func=pbincli.actions.send)
get_parser = subparsers.add_parser("get", description="Get data from PrivateBin instance", usage=""" # a get command
%(prog)s pasteid#password""" 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("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("-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) get_parser.set_defaults(func=pbincli.actions.get)
delete_parser = subparsers.add_parser("delete", description="Delete paste from PrivateBin instance using token", usage=""" # a delete command
%(prog)s --paste aabb --token aabbcc""" 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("-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.add_argument("-d", "--debug", default=False, action="store_true", help="enable debug")
delete_parser.set_defaults(func=pbincli.actions.delete) delete_parser.set_defaults(func=pbincli.actions.delete)
# parse arguments # parse arguments
args = parser.parse_args() args = parser.parse_args()
CONFIG = {"server": "https://paste.i2pd.xyz/", CONFIG = {
"proxy": None} '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: for p in CONFIG_PATHS:
if os.path.exists(p): if os.path.exists(p):
@@ -72,15 +101,21 @@ def main():
for key in CONFIG.keys(): for key in CONFIG.keys():
var = "PRIVATEBIN_{}".format(key.upper()) var = "PRIVATEBIN_{}".format(key.upper())
if var in os.environ: CONFIG[key] = os.getenv(var) 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"): if hasattr(args, "func"):
try: try:
args.func(args, api_client) args.func(args, api_client, settings=CONFIG)
except PBinCLIException as pe: except PBinCLIException as pe:
print("PBinCLI error: {}".format(pe)) raise PBinCLIException("error: {}".format(pe))
sys.exit(1)
else: else:
parser.print_help() parser.print_help()

290
pbincli/format.py Normal file
View 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)

View File

@@ -1,50 +1,36 @@
import json, os import json, ntpath, os, sys
class PBinCLIException(Exception): class PBinCLIException(Exception):
pass 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): def check_readable(f):
# Checks if path exists and readable # Checks if path exists and readable
if not os.path.exists(f) or not os.access(f, os.R_OK): 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): def check_writable(f):
# Checks if path is writable # Checks if path is writable
if not os.access(os.path.dirname(f) or ".", os.W_OK): 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_encode(s):
def json_load_byteified(file_handle): return json.dumps(s, separators=(',',':')).encode()
return _byteify(
json.load(file_handle, object_hook=_byteify),
ignore_dicts=True
)
def json_loads_byteified(json_text): def validate_url(s):
return _byteify( if not s.endswith('/'):
json.loads(json_text, object_hook=_byteify), s = s + "/"
ignore_dicts=True return s
)
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

View File

@@ -1,3 +1,4 @@
pycryptodome pycryptodome
sjcl sjcl
base58
requests requests

View File

@@ -1,8 +1,9 @@
#!/usr/bin/env python #!/usr/bin/env python
from setuptools import setup 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() long_description = readme.read()
with open("requirements.txt") as f: with open("requirements.txt") as f:
@@ -10,19 +11,23 @@ with open("requirements.txt") as f:
setup( setup(
name='PBinCLI', name='PBinCLI',
version='0.1', version=pbincli_version,
description='PrivateBin client for command line', description='PrivateBin client for command line',
long_description=long_description, long_description=long_description,
long_description_content_type='text/x-rst',
author='R4SAS', author='R4SAS',
author_email='r4sas@i2pmail.org', author_email='r4sas@i2pmail.org',
url='https://github.com/r4sas/PBinCLI', url='https://github.com/r4sas/PBinCLI',
keywords='privatebin', keywords='privatebin cryptography security',
license='DWTFYWWI', license='MIT',
classifiers=[ classifiers=[
'Development Status :: 4 - Beta', 'Development Status :: 5 - Production/Stable',
'License :: Other/Proprietary License',
'Programming Language :: Python :: 3',
'Environment :: Console', 'Environment :: Console',
'Intended Audience :: End Users/Desktop',
'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 3',
'Topic :: Security :: Cryptography',
'Topic :: Utilities',
], ],
packages=['pbincli'], packages=['pbincli'],
install_requires=install_requires, install_requires=install_requires,