Produits Entr'ouvert » Passerelle

    python-pycryptodome,

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2020-03-04 10:25

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

import passerelle.apps.cryptor.models

import uuid

class Migration(migrations.Migration):

    initial = True

    dependencies = [

        ('base', '0016_auto_20191002_1443'),

    ]

    operations = [

        migrations.CreateModel(

            name='CryptedFile',

            fields=[

                ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),

                ('filename', models.CharField(max_length=512)),

                ('content_type', models.CharField(max_length=128)),

                ('creation_timestamp', models.DateTimeField(auto_now_add=True)),

            ],

        ),

        migrations.CreateModel(

            name='Cryptor',

            fields=[

                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),

                ('title', models.CharField(max_length=50, verbose_name='Title')),

                ('slug', models.SlugField(unique=True, verbose_name='Identifier')),

                ('description', models.TextField(verbose_name='Description')),

                ('public_key', models.TextField(blank=True, validators=[passerelle.apps.cryptor.models.validate_rsa_key], verbose_name='Encryption RSA public key (PEM format)')),

                ('private_key', models.TextField(blank=True, validators=[passerelle.apps.cryptor.models.validate_rsa_key], verbose_name='Decryption RSA private key (PEM format)')),

                ('redirect_url_base', models.URLField(blank=True, help_text='Base URL for redirect, empty for local', max_length=256, verbose_name='Base URL of decrypt system')),

                ('users', models.ManyToManyField(blank=True, related_name='_cryptor_users_+', related_query_name='+', to='base.ApiUser')),

            ],

            options={

                'verbose_name': 'Encryption / Decryption',

            },

        ),

        migrations.AddField(

            model_name='cryptedfile',

            name='resource',

            field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='cryptor.Cryptor'),

        ),

    ]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2020  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import base64

import binascii

import json

import os

from uuid import uuid4

from Cryptodome.PublicKey import RSA

from Cryptodome.Random import get_random_bytes

from Cryptodome.Cipher import AES, PKCS1_OAEP

from django.core.exceptions import ValidationError

from django.core.files.storage import default_storage

from django.db import models

from django.http import HttpResponse

from django.utils.six.moves.urllib_parse import urljoin

from django.utils.translation import ugettext_lazy as _

from passerelle.base.models import BaseResource

from passerelle.utils.api import endpoint

from passerelle.utils.files import atomic_write

from passerelle.utils.jsonresponse import APIError

FILE_SCHEMA = {

    "$schema": "http://json-schema.org/draft-04/schema#",

    "title": "File to encrypt",

    "description": "",

    "type": "object",

    "required": ["file"],

    "properties": {

        "file": {

            "type": "object",

            "required": ["filename", "content_type", "content"],

            "properties": {

                "filename": {"type": "string"},

                "content_type": {"type": "string"},

                "content": {"type": "string"},

            }

        }

    }

}

# encrypt and decrypt are borrowed from

# https://www.pycryptodome.org/en/latest/src/examples.html#encrypt-data-with-rsa

def write_encrypt(out_file, data, key_pem):

    public_key = RSA.import_key(key_pem)

    session_key = get_random_bytes(16)

    # Encrypt the session key with the public RSA key

    cipher_rsa = PKCS1_OAEP.new(public_key)

    enc_session_key = cipher_rsa.encrypt(session_key)

    # Encrypt the data with the AES session key

    cipher_aes = AES.new(session_key, AES.MODE_EAX)

    ciphertext, tag = cipher_aes.encrypt_and_digest(data)

    # Store in out_file

    out_file.write(enc_session_key)

    out_file.write(cipher_aes.nonce)

    out_file.write(tag)

    out_file.write(ciphertext)

def read_decrypt(in_file, key_pem):

    private_key = RSA.import_key(key_pem)

    # Get crypt elements from in_file

    enc_session_key = in_file.read(private_key.size_in_bytes())

    nonce = in_file.read(16)

    tag = in_file.read(16)

    ciphertext = in_file.read()

    # Decrypt the session key with the private RSA key

    cipher_rsa = PKCS1_OAEP.new(private_key)

    session_key = cipher_rsa.decrypt(enc_session_key)

    # Decrypt the data with the AES session key

    cipher_aes = AES.new(session_key, AES.MODE_EAX, nonce)

    decrypted = cipher_aes.decrypt_and_verify(ciphertext, tag)

    return decrypted

def makedir(dir_name):

    if not os.path.exists(dir_name):

        if default_storage.directory_permissions_mode:

            d_umask = os.umask(0)

            try:

                os.makedirs(dir_name, mode=default_storage.directory_permissions_mode)

            except OSError:

                pass

            finally:

                os.umask(d_umask)

        else:

            os.makedirs(dir_name)

def validate_rsa_key(key):

    try:

        RSA.import_key(key)

    except ValueError as ex:

        raise ValidationError(_('invalid RSA key (%s)') % ex)

class Cryptor(BaseResource):

    public_key = models.TextField(blank=True,

                                  verbose_name=_('Encryption RSA public key (PEM format)'),

                                  validators=[validate_rsa_key])

    private_key = models.TextField(blank=True,

                                   verbose_name=_('Decryption RSA private key (PEM format)'),

                                   validators=[validate_rsa_key])

    redirect_url_base = models.URLField(max_length=256, blank=True,

                                        verbose_name=_('Base URL of decrypt system'),

                                        help_text=_('Base URL for redirect, empty for local'))

    category = _('Misc')

    class Meta:

        verbose_name = _('Encryption / Decryption')

    def get_redirect_url_base_display(self):

        if self.redirect_url_base:

            return _('defined')  # don't show it, can be sensitive

        return _('this file-decrypt endpoint')

    def get_filename(self, uuid, create=False):

        dirname = os.path.join(default_storage.path(self.get_connector_slug()),

                               self.slug, uuid[0:2], uuid[2:4])

        if create:

            makedir(dirname)

        filename = os.path.join(dirname, uuid)

        return filename

    @endpoint(name='file-encrypt', perm='can_encrypt',

              description=_('Encrypt a file'),

              post={

                  'description': _('File to encrypt'),

                  'request_body': {'schema': {'application/json': FILE_SCHEMA}}

              })

    def file_encrypt(self, request, post_data):

        if not self.public_key:

            raise APIError('missing public key')

        try:

            data = base64.b64decode(post_data['file']['content'])

        except (TypeError, binascii.Error):

            raise APIError('invalid base64 string', http_status=400)

        filename = post_data['file']['filename']

        content_type = post_data['file']['content_type']

        cfile = CryptedFile(resource=self, filename=filename, content_type=content_type)

        cfile.save()

        uuid = str(cfile.uuid)  # get string representation of UUID object

        if self.redirect_url_base:

            redirect_url_base = self.redirect_url_base

        else:

            redirect_url_base = request.build_absolute_uri('%sfile-decrypt/' % (

                self.get_absolute_url(),))

        redirect_url = urljoin(redirect_url_base, uuid)

        content_filename = self.get_filename(uuid, create=True)

        metadata_filename = '%s.json' % content_filename

        metadata = {

            'filename': cfile.filename,

            'content_type': cfile.content_type,

            'creation_timestamp': cfile.creation_timestamp.isoformat(),

            'redirect_url': redirect_url,

        }

        tmp_dir = os.path.join(default_storage.path(self.get_connector_slug()), self.slug, 'tmp')

        with atomic_write(content_filename, dir=tmp_dir) as fd:

            write_encrypt(fd, data, self.public_key)

        with atomic_write(metadata_filename, dir=tmp_dir, mode='w') as fd:

            json.dump(metadata, fd, indent=2)

        return {'data': metadata}

    @endpoint(name='file-decrypt', perm='can_decrypt',

              description=_('Decrypt a file'),

              pattern=r'(?P<uuid>[\w-]+)$',

              example_pattern='{uuid}/',

              parameters={

                  'uuid': {

                      'description': _('File identifier'),

                      'example_value': '12345678-abcd-4321-abcd-123456789012',

                  },

              })

    def file_decrypt(self, request, uuid):

        if not self.private_key:

            raise APIError('missing private key')

        content_filename = self.get_filename(uuid, create=False)

        metadata_filename = '%s.json' % content_filename

        if not os.path.exists(metadata_filename):

            raise APIError('unknown uuid', http_status=404)

        content = read_decrypt(open(content_filename, 'rb'), self.private_key)

        metadata = json.load(open(metadata_filename, 'r'))

        filename = metadata.get('filename')

        content_type = metadata.get('content_type')

        response = HttpResponse(content_type=content_type)

        response['Content-Disposition'] = 'inline; filename="%s"' % filename

        response.write(content)

        return response

class CryptedFile(models.Model):

    resource = models.ForeignKey(Cryptor, on_delete=models.PROTECT)

    uuid = models.UUIDField(primary_key=True, default=uuid4, editable=False)

    filename = models.CharField(max_length=512, blank=False)

    content_type = models.CharField(max_length=128)

    creation_timestamp = models.DateTimeField(auto_now_add=True)

    'passerelle.apps.cryptor',

li.connector.cryptor a::before {

	content: "\f023";  /* lock */

}

            'pycryptodomex',

# -*- coding: utf-8 -*-

import base64

import pytest

from django.contrib.contenttypes.models import ContentType

from django.core.exceptions import ValidationError

from django.utils.encoding import force_text

from passerelle.apps.cryptor.models import Cryptor, CryptedFile

from passerelle.base.models import ApiUser, AccessRight

import utils

PUBLIC_KEY = '''-----BEGIN CERTIFICATE-----

MIICyjCCAbKgAwIBAgIUQQzM2eFYF+LpUR3t2euAjZAwLCEwDQYJKoZIhvcNAQEL

BQAwDzENMAsGA1UEAwwEemVwbzAeFw0xOTA2MjIyMjMxMTVaFw0yOTA2MTkyMjMx

MTVaMA8xDTALBgNVBAMMBHplcG8wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK

AoIBAQC8BM3xDylze0bOm76IjidyhmFqJlnRvcpbeZVTM7r3qYOHqXFG7/GZL4yd

2bW5eL6TCUT3gLEgegGYGPwCkGPd1cq9h+2M7zvolGToRCvrBpxH5Vu6iRkEYyWN

yPhc02EUqYlz1FBBYRgyYHQ4jy0vsPH55g536OKLI4rVykszjwD9p0Kh+T2I1D9Z

bHyA6s0C8goUFZG7kvasFRIXTTPgUBGSnEN/VPSD5vM94Oj5K4t6P9GHd32Jo2O6

E5jAHbPR7I4nRBCJuxJEbfpmsaOuMkGQ5rMulk6LXvRZiT9UDCDem1k6uF6tJkZR

g+Uh5V4ZLCzP7sSHcRN2ftWZqAr7AgMBAAGjHjAcMAkGA1UdEwQCMAAwDwYDVR0R

BAgwBoIEemVwbzANBgkqhkiG9w0BAQsFAAOCAQEANo54TMbOR5Isd4lix87EM0N+

8kxovCLin/szK4+fGfnr0fCUswkhoZ3y6xnmXFX4S2IGLU8bTl+eQVg04VM/7Gg2

LvBTtiBmGESbiSaC1fS+DbPBjp1NBpfwbiQFEuQfRMS6ejeF1YMS8Oy9PpeujHDT

4cX0kPH2GkqOGtpAdKoOD5XzT3yu5IHv7/AWpl8LZ5hr3f1RbzfIzJ19oC5NDXIR

d1XsjuCCHFbCyfnDmuZuQaGCTCm9f4Z8Ynum6hmSSzNvy3YJLRKXEYYLB3+l2V+t

SCzQiVzqDeKZnAChvJRditvcKdG36TPHREyCPgkpUTmi0gEjDZDPyBmXhHXHYA==

-----END CERTIFICATE-----

'''

PRIVATE_KEY = '''-----BEGIN PRIVATE KEY-----

MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC8BM3xDylze0bO

m76IjidyhmFqJlnRvcpbeZVTM7r3qYOHqXFG7/GZL4yd2bW5eL6TCUT3gLEgegGY

GPwCkGPd1cq9h+2M7zvolGToRCvrBpxH5Vu6iRkEYyWNyPhc02EUqYlz1FBBYRgy

YHQ4jy0vsPH55g536OKLI4rVykszjwD9p0Kh+T2I1D9ZbHyA6s0C8goUFZG7kvas

FRIXTTPgUBGSnEN/VPSD5vM94Oj5K4t6P9GHd32Jo2O6E5jAHbPR7I4nRBCJuxJE

bfpmsaOuMkGQ5rMulk6LXvRZiT9UDCDem1k6uF6tJkZRg+Uh5V4ZLCzP7sSHcRN2

ftWZqAr7AgMBAAECggEAUQsdHhA0BNQZdEtLuJ7VwBbOfKvlQXQ2enGQ/RkqOUC3

Mk3GRxZ8JFSLnyrNmxHBy61OLgUp1F7iuwXh8tT8Rw21YzbpHTutrhXw3PEtoRPr

X04s2N3pi6uU72W2MITorrhZSDU3FsdcX7KVxh9pEcqKsvYIPIWEyQbb/EVDXwhC

K4TAsmmhsGaU/BB0WHkbzU5KlZqYQHfnoj5pmTLoeYj81Z8D5T9fceImcyuWLl8t

OHscbAjNpkS1X2vYNwMLhCAM7YhE63RjFo2G5fxx64wdgs+mllR0PCc3Nli+JPWE

LQ/KmVMPY6JH506WbZkEVgb9Gfuj6yASpu5zJxf3MQKBgQDwktbzKmK3zwSNwLdx

zZy4AnQb2PNKuryYts9R+nKqWnE5vJwekBIV9vsASK0Aoh4UWP9cUjQV3gZH3HIR

9xP8nJwpOLU7c/wPP81HUl7nf68MVP5OzemHY/78fYR8T3Jf/uFdW3AOQrCBjGIV

Zbjb5SisqCS2ODc7DSuqzliBMwKBgQDIEz7nhXyMilTlgiWw7iRm9RKWykVSjEuR

gxHAqPOkg/HxacfDYs/O+qteSdW/V7zzZ2QHMyr52BaL5sEz0lC7W3O1nfTS94VM

YK+MakRBImC5uZvpea+30vJbVLODnKskhsWF8aCqLi5KJvkR5aE0OySKXBur7Czq

X2/gK6jfGQKBgAbU1J/BE16O1V1FHLBxm0KqZyunRHlZxiM8BbUZPIpT2SU/kttX

UfwnsEb4yVjcQahoQpAXkX0RefIuc1rJPlsNA240Owk+KOkx8Z1V3HYMbScXfsU0

Ga6Li2EWG14AT4okTbf98bel8ycqmlprMg2kezwz5h76h674l8XY6DB7AoGBAKJL

Aka5gBtchpsZJEvOENc3Spnof60DQrVJVZgrNF+p7BMA1FsIhzsFGQdF603n9MyY

fIpelijOgROA3g2UN4qTF1wmQhbzUzxuXVgQR0dyhHWDOxZ7b+8z/QXawjcrWaQq

coVBSCtjhIb/8B/1XftJUk2tg4DE9nYzbkOwBq7ZAoGAPF1KSY9TzY7g8gmQzYuY

+CCHM3mR9PjSHhJS5VWTwGfw3zZpLptwxzmJAoi1DyrJVhBP17ADVitYK4GquiSB

z5aZ2AnUBc/xueO2ixL3ROOXYAeakrRAQ38G13ibYe2dQpv6/CTsZJOttnCErn54

3k2Y/kDV+c1uNbzyPiK2qaM=

-----END PRIVATE KEY-----

'''

@pytest.fixture

def cryptor(db):

    return Cryptor.objects.create(slug='test',

                                  private_key=PRIVATE_KEY,

                                  public_key=PUBLIC_KEY)

def test_cryptor_bad_keys(db):

    bad1 = Cryptor(slug='bad1', title='t', description='d', private_key='badkey')

    with pytest.raises(ValidationError) as e1:

        bad1.full_clean()

    assert e1.value.messages == ['invalid RSA key (RSA key format is not supported)']

    bad2 = Cryptor(slug='bad2', title='t', description='d', public_key='badkey')

    with pytest.raises(ValidationError) as e2:

        bad2.full_clean()

    assert e2.value.messages == ['invalid RSA key (RSA key format is not supported)']

def test_cryptor_restricted_access(app, cryptor):

    endpoint = utils.generic_endpoint_url('cryptor', 'file-encrypt', slug=cryptor.slug)

    assert endpoint == '/cryptor/test/file-encrypt'

    resp = app.get(endpoint, status=405)

    resp = app.post_json(endpoint, params={"foo": "bar"}, status=403)

    assert resp.json['err'] == 1

    assert 'PermissionDenied' in resp.json['err_class']

    endpoint = utils.generic_endpoint_url('cryptor', 'file-decrypt', slug=cryptor.slug) + '/uuid'

    assert endpoint == '/cryptor/test/file-decrypt/uuid'

    resp = app.post_json(endpoint, params={"foo": "bar"}, status=405)

    resp = app.get(endpoint, status=403)

    assert resp.json['err'] == 1

    assert 'PermissionDenied' in resp.json['err_class']

def test_cryptor_bad_requests(app, cryptor):

    # full opened access

    api = ApiUser.objects.create(username='all', keytype='', key='')

    obj_type = ContentType.objects.get_for_model(cryptor)

    AccessRight.objects.create(codename='can_encrypt', apiuser=api, resource_type=obj_type,

                               resource_pk=cryptor.pk)

    AccessRight.objects.create(codename='can_decrypt', apiuser=api, resource_type=obj_type,

                               resource_pk=cryptor.pk)

    endpoint = utils.generic_endpoint_url('cryptor', 'file-encrypt', slug=cryptor.slug)

    for bad_payload in ('error',

                        {"foo": "bar"},

                        ["not", "a", "dict"],

                        {"file": {"filename": "f", "content_type": "ct"}},

                        {"file": {"filename": "f", "content_type": "ct", "content": None}},

                        {"file": {"filename": "f", "content_type": "ct", "content": "NotBase64"}},

                        ):

        resp = app.post_json(endpoint, params=bad_payload, status=400)

        assert resp.json['err'] == 1

    endpoint = utils.generic_endpoint_url('cryptor', 'file-decrypt', slug=cryptor.slug)

    endpoint = endpoint + '/bad-uuid'

    resp = app.get(endpoint, status=404)

def test_cryptor_encrypt_decrypt(app, cryptor):

    api = ApiUser.objects.create(username='all', keytype='', key='')

    obj_type = ContentType.objects.get_for_model(cryptor)

    AccessRight.objects.create(codename='can_encrypt', apiuser=api, resource_type=obj_type,

                               resource_pk=cryptor.pk)

    AccessRight.objects.create(codename='can_decrypt', apiuser=api, resource_type=obj_type,

                               resource_pk=cryptor.pk)

    # encrypt

    endpoint = utils.generic_endpoint_url('cryptor', 'file-encrypt', slug=cryptor.slug)

    content = force_text(base64.b64encode(b'this is foo and bar'))

    payload = {"file": {"filename": "foo.txt", "content_type": "text/plain", "content": content}}

    resp = app.post_json(endpoint, params=payload, status=200)

    assert resp.json['err'] == 0

    assert CryptedFile.objects.count() == 1

    cfile = CryptedFile.objects.first()

    assert resp.json['data']['redirect_url'].endswith(

            '/cryptor/%s/file-decrypt/%s' % (cryptor.slug, cfile.uuid))

    cfile.delete()

    # encrypt with another redirect url

    cryptor.redirect_url_base = 'https://foo/bar/'

    cryptor.save()

    resp = app.post_json(endpoint, params=payload, status=200)

    assert resp.json['err'] == 0

    assert CryptedFile.objects.count() == 1

    cfile = CryptedFile.objects.first()

    assert resp.json['data']['redirect_url'] == 'https://foo/bar/%s' % cfile.uuid

    # remove public key = cannot encrypt

    cryptor.public_key = ''

    cryptor.save()

    resp = app.post_json(endpoint, params=payload, status=200)

    assert resp.json['err'] == 1

    assert resp.json['err_desc'] == 'missing public key'

    # decrypt

    endpoint = utils.generic_endpoint_url('cryptor', 'file-decrypt', slug=cryptor.slug)

    endpoint = endpoint + '/' + str(cfile.uuid)

    resp = app.get(endpoint, status=200)

    assert resp.content_type == 'text/plain'

    assert resp.content == b'this is foo and bar'

    assert resp.headers['Content-Disposition'] == 'inline; filename="foo.txt"'

    # remove all files does not remove data+metadata(json) files: decrypt still works

    CryptedFile.objects.all().delete()

    assert CryptedFile.objects.count() == 0

    resp = app.get(endpoint, status=200)

    assert resp.content_type == 'text/plain'

    assert resp.content == b'this is foo and bar'

    assert resp.headers['Content-Disposition'] == 'inline; filename="foo.txt"'

    # remove private key = cannot decrypt

    cryptor.private_key = ''

    cryptor.save()

    endpoint = endpoint + '/' + str(cfile.uuid)

    resp = app.get(endpoint, status=200)

    assert resp.json['err'] == 1

    assert resp.json['err_desc'] == 'missing private key'