Produits Entr'ouvert » Passerelle

{% load i18n %}

<div class="sftp-widget-url">

    <label for="{{ widget.subwidgets.0.attrs.id }}">{% trans "URL" %}</label>

    {% include widget.subwidgets.0.template_name with widget=widget.subwidgets.0 %}

</div>

<div class="sftp-widget-private-key">

    <label for="{{ widget.subwidgets.1.attrs.id }}">{% trans "SSH private key" %}</label>

    {% include widget.subwidgets.1.template_name with widget=widget.subwidgets.1 %}

</div>

<div class="sftp-widget-private-key-text">

    {% include widget.subwidgets.2.template_name with widget=widget.subwidgets.2 %}

</div>

<div class="sftp-widget-private-key-password">

    <label for="{{ widget.subwidgets.3.attrs.id }}">{% trans "SSH private key password" %}</label>

    {% include widget.subwidgets.3.template_name with widget=widget.subwidgets.3 %}

</div>

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2019 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 os

import re

import io

import json

import contextlib

from django import forms

from django.core import validators

from django.db import models

from django.utils import six

from django.utils.translation import ugettext_lazy as _

from django.utils.six.moves.urllib import parse as urlparse

from django.utils.encoding import force_bytes

import paramiko

from paramiko.dsskey import DSSKey

from paramiko.ecdsakey import ECDSAKey

from paramiko.ed25519key import Ed25519Key

from paramiko.rsakey import RSAKey

def _load_private_key(content_or_file, password=None):

    if not hasattr(content_or_file, 'read'):

        content_or_file = io.BytesIO(force_bytes(content_or_file))

    for pkey_class in RSAKey, DSSKey, Ed25519Key, ECDSAKey:

        try:

            return pkey_class.from_private_key(

                content_or_file,

                password=password)

        except paramiko.PasswordRequiredException:

            raise

        except paramiko.SSHException:

            pass

@six.python_2_unicode_compatible

class SFTP(object):

    def __init__(self, url, private_key_content=None, private_key_password=None):

        self.url = url

        parsed = urlparse.urlparse(url)

        if not parsed.scheme == 'sftp':

            raise ValueError('invalid scheme %s' % parsed.scheme)

        if not parsed.hostname:

            raise ValueError('missing hostname')

        self.username = parsed.username or None

        self.password = parsed.password or None

        self.hostname = parsed.hostname

        self.port = parsed.port or 22

        self.path = parsed.path.strip('/')

        self.private_key_content = private_key_content

        self.private_key_password = private_key_password

        if private_key_content:

            self.private_key = _load_private_key(private_key_content, private_key_password)

        else:

            self.private_key = None

        self._client = None

        self._transport = None

    def __json__(self):

        return {

            'url': self.url,

            'private_key_content': self.private_key_content,

            'private_key_password': self.private_key_password,

        }

    def __str__(self):

        return re.sub(r'://([^/]*:[^/]*?)@', '://***:***@', self.url)

    # Paramiko can hang processes if not closed, it's important to use it as a

    # contextmanager

    @contextlib.contextmanager

    def client(self):

        ssh = paramiko.SSHClient()

        try:

            ssh.set_missing_host_key_policy(paramiko.client.AutoAddPolicy)

            ssh.connect(

                hostname=self.hostname,

                port=self.port,

                timeout=5,

                pkey=self.private_key,

                look_for_keys=False,

                allow_agent=False,

                username=self.username,

                password=self.password)

            client = ssh.open_sftp()

            try:

                if self.path:

                    client.chdir(self.path)

                    base_cwd = str(client._cwd)

                    old_adjust_cwd = client._adjust_cwd

                    def _adjust_cwd(path):

                        path = old_adjust_cwd(path)

                        if not os.path.normpath(path).startswith(base_cwd):

                            raise ValueError('all paths must be under base path %s: %s' % (base_cwd, path))

                        return path

                    client._adjust_cwd = _adjust_cwd

                yield client

            finally:

                client.close()

        finally:

            ssh.close()

class SFTPURLField(forms.URLField):

    default_validators = [validators.URLValidator(schemes=['sftp'])]

class SFTPWidget(forms.MultiWidget):

    template_name = 'passerelle/widgets/sftp.html'

    def __init__(self, **kwargs):

        widgets = [

            forms.TextInput,

            forms.FileInput,

            forms.Textarea,

            forms.TextInput,

        ]

        super(SFTPWidget, self).__init__(widgets=widgets, **kwargs)

    def decompress(self, value):

        if not value:

            return [None, None, None, None]

        if hasattr(value, '__json__'):

            value = value.__json__()

        return [

            value['url'],

            None,

            value.get('private_key_content'),

            value.get('private_key_password'),

        ]

    # XXX: bug in Django https://code.djangoproject.com/ticket/29205

    # required_attribute is initialized from the parent.field required

    # attribute and not from each sub-field attribute

    def use_required_attribute(self, initial):

        return False

class SFTPFormField(forms.MultiValueField):

    widget = SFTPWidget

    def __init__(self, **kwargs):

        fields = [

            SFTPURLField(),

            forms.FileField(required=False),

            forms.CharField(required=False),

            forms.CharField(required=False),

        ]

        super(SFTPFormField, self).__init__(

            fields=fields,

            require_all_fields=False, **kwargs)

#    def clean(self, value):

#        import pdb

#        pdb.set_trace()

#        return super(SFTPFormField, self).clean(value)

    def compress(self, data_list):

        url, private_key_file, private_key_content, private_key_password = data_list

        if private_key_file:

            private_key_content = private_key_file.read().decode('ascii')

        if private_key_content:

            try:

                pkey = _load_private_key(private_key_content, private_key_password)

            except paramiko.PasswordRequiredException:

                raise forms.ValidationError(_('SSH private key needs a password'))

            if not pkey:

                raise forms.ValidationError(_('SSH private key invalid'))

        return SFTP(

            url=url,

            private_key_content=private_key_content,

            private_key_password=private_key_password)

class SFTPField(models.Field):

    description = 'A SFTP connection'

    def __init__(self, **kwargs):

        kwargs.setdefault('default', None)

        super(SFTPField, self).__init__(**kwargs)

    def get_internal_type(self):

        return 'TextField'

    def from_db_value(self, value, *args, **kwargs):

        return self.to_python(value)

    def to_python(self, value):

        if not value:

            return None

        if isinstance(value, SFTP):

            return value

        if isinstance(value, six.string_types):

            return SFTP(url=value)

        return SFTP(**json.loads(value))

    def get_prep_value(self, value):

        if not value:

            return ''

        return json.dumps(value.__json__())

    def formfield(self, **kwargs):

        defaults = {

            'form_class': SFTPFormField,

        }

        defaults.update(**kwargs)

        return super(SFTPField, self).formfield(**defaults)

            'paramiko',

-----BEGIN RSA PRIVATE KEY-----

MIIEpAIBAAKCAQEAqH/MTAXMKqz3uQDcliK9udOXlGNUlPN/gVMdxsw0kwmrKdrW

svuIa3UhrHI/6bPnTTniKvXmccAIm8aAdEg8rAgcksAada6qUYbc0aZbR0WyC+3/

fvaLbD9WutD05frrIEwjIUkcXavWSlaHrEzV2T3mGGI8amqR//bjwksRkClx2vvW

apc5FkzGtxKsc0tA6ZwyA3riqCeEVwOx+ezV8B/631q62h6AAt0j4cb+THzD1t3w

6KAsxt72lNw6Bp55uR5Uubin3OpijEVL4Gruy1sSJO6pQ5YHpDFTS9ZAHLLldUea

w+hNha+fGM/EvQJEGZsZSA/urEAuC6Xly8T/JwIDAQABAoIBAGjMe4s4+9/7DmQB

VjEG0IvYP2mqUfwGamJMCLQRZA2jsNJqaqiNay6yfkwcDwZSv2S3wKRJppdPAcup

LVGlcB7rOKJJWuugxAvK3mKCnjj47yEeWI9l1hdwWYf92KOFaWIAGMVmDH9yFejM

YrvWWhcwuYCm8L6bI81YiBXazMSlD+UXB9WOityCc5DayXhTvKiP6mzVQHJJ4p4o

q/LtBI4jOI7ZEnlI+lYNBvCfZLaPIi7vvLpOXG263ZRlwYr+IjED+7Ex6zjL3YEd

VFfp2MX5uY3IsXmjyhHokm2f2JRH1JtigjkRysT4AYHk76XktrsczfY6Hgs6Lwz8

n6/o2sECgYEA09ErhKlObtWfWdp0HXd/HRubvBQBQ+oYPxNcb6QdWOaXzG8jDBuN

bxOFXdqknkWzG6pNlurjVW4njeoovts4Wmza5U4Ju9IL2lrUNXqRkopbdcPCw0v6

kdJqbui3DwOA/gmphO802jCXu+xASSdoVK8/sCC/itOcAzi8E0YUHCECgYEAy6V+

3ivJczRT9xM7hOJA1pJDpLbbuE6XTQBn9QfoP90zAC6Tbr5k7dh0Uyc4JcTOPK7C

aMmr8ScRL9UxZKsDwxFQPJwz4/R3VCcmJt0j0vgilwm2ujiqWmgqgvM4YBg+dzlU

J05PenMgzYAp9HD9a8B6d6lOvioF0jx1nbZa8kcCgYEAqJhhDyLDryyRva9HpPys

TLrg5n710tzNl8cNWD9ErLI+ORZsywJTPQpIqT+Sr/fCbE7Nm0Yy1JjtGuQ6sk9D

N5ZVVRccYEb78D1Dk52PqRg/XCkJKPGc69yTotvQeT7MuWdvasQLSXBMFeQh9xhK

zrz+8G3gh9uO3nGWIbEx6IECgYBUW73uMp1Eh8ywcNsa9M5/FB/JP6ZM9uFeGGj3

68qdiffyf1i7a0tL63pkZ76uhpQYNxx5Y/FB+Dj6Y4oOdXkdeTKPqPUl3MMBrSX0

u253mipZ/sAe7BJFWRkjHbWguOpHYQwnLB1oUACqoAjBJX0VAaq5nvzrcWTv7fOa

3UtXSQKBgQC8un5szEXQth2viUe5P7FyMsz+34XRKL7qCaCv1/4kGgPhlSbkYHao

PZ8u5hue/hnmUfpbXNJfc+zazTIojuZ047Z5rJYpKCfs5JozuGOGVOVIf5hunWfJ

FU3c4umawKvh5tUUeyrXilzEWnVECGYDyzj9BumGT+YkG94yll7iZw==

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

-----BEGIN RSA PRIVATE KEY-----

Proc-Type: 4,ENCRYPTED

DEK-Info: AES-128-CBC,AE8F07BCBA597172BBC8D034B3C24C30

eg/EgNvQzSeH+W7GP23U2XIbzqg3UQgnSqi+3GbZcZUnqYmsmChiMZy+Wb0b1BIY

QOFLvOkaiP3hSnfFvx8oSD7/J2h633uSvhtGrq9wIMVUCXLLgBL2k2pzbWHPz1WG

6eJRAdWQ8c7jCCVkLdFf077ebJXL3tCU/1gP7RJ2wLhkfRXtWARgYPrmw2+ZY6l1

tre1O0AS+yjI6ZAsVjPCPqq9WrP4CggtQ3W33/AERGrAWi9PJZFjQUE0cMj6J+R0

uciudN+PgO3SRKc+uWX3YrUc5zb+4SMvRO2uMiULyVMahJnnpHSO5qB+dcn6vjcU

F6ogpfyJpNgm4dCvEwtEiuva1HKlzCQRpWVlHhsRM6l7/faxON8QNND2ZKgO5XS6

QFvBbshOLssM7XQQIfdk4yNjx7lWA63W0E+t6sQqTMTeLbWePIq5IlvazFlSqkIS

I2B3e0Dd3emStwhZhEDncPhysh0P0WOCqnXiql3uGZGkL6bl47qRD2Ft0DnWoNoJ

0LUTClhInvgi9K50iMup/pq4HN8bo3fAgAcMePF3AqktFTzyIzKK7/pt0jjdX+7N

MBU9YkrDA3BzwupT71M/hQJIoqGBD68RGFR2Wmxc73xy6Pmv5DDv0e/fGpo0dnHC

IuhKMs24uRn85OWIGUi+GAbwp9OlJ6Vlh59ZAl4KVSySAnaGCZiAHfWpkWrGMlkd

wMi1V/L/NGXyzaereK4PUT3IKPFglRr2bTfwrhX83zO5fU4vMuMy22KNQpoX9agG

J2IkyhMsd56x2JmAhQKLi3rLagzEdx2a4pRJyoU8n95/NhLYnPBBl73weJKtZj/t

8KNhqjgerPPXdgYua2TE+4e81XrYWhbqdepG0GiYZ23XBIAIor2jgGjswENUw2TZ

TEFrs9MAx/0r1isz4UurXvyA4IX8TANfJzEqZwimNcn9Eehj6drnvO5GRKCbVRhe

p0mqMqXnTZGdJyT4EXSNnbRxW5jkaLbDSaiJ+6U0fLawqGRR0VazT4JlOAVz8i3i

BJxRpYZ/PvRXevh1QrUpgS/wds7+b6nyf4XgV1RsvRS2uuyqx6smE5d+sW9qVZsU

kuTgI99zr7JSXUO1IDqfBXldnfMVJyGvYMGXdd3UGOis23wDlAd61H6YJAwp75oR

pOv45Jn8JVQ0w1yFZXTZz2PyDztVpH46RQLxHMPjKber7qN8Lfhc97+01tisV0UB

k6cWuAXZo61uBr35jr2BRHS1xnrK8Ul4sJPq9LUUaJU2+oZeKNmytTbMRel6w156

d64QU75OsT5tZHKKV/NQDLuAWY1ZB+useZNHSkqz860zNzRA+gyb6pLL88Z/Q2YS

z0OJMXNHxBZhNKk6SPZQTIo6Q0bn8dxY8BwCLSwJGi0NMEugm5f+9+GTjYNplGTm

z0itYZ8dyhk+rc/48eUg/ctCGM7FbvBaLGkBGqSUrPDFzZjvqASLtKD2IqUyU1bW

EDlz9PHSX+SGP1A3Y5IKd1L3dbq7lO0SlJnkAOReaagjwNJ+f3dxkDikkBRO+/SB

V7zuK8XIc1D0Hy13AGdGsiiBF8BXCW7hrhnY+7hKHvtw4nljQ5CGKWhPBauohMdG

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

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2019 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 os

import pytest

from django.core.files.uploadedfile import SimpleUploadedFile

from django.db import models

from passerelle.utils.sftp import SFTP, SFTPFormField, SFTPField

@pytest.fixture

def ssh_key():

    with open(

        os.path.join(

            os.path.dirname(__file__),

            'ssh_key'), 'rb') as fd:

        yield fd.read()

@pytest.fixture

def ssh_key_with_password():

    with open(

        os.path.join(

            os.path.dirname(__file__),

            'ssh_key_with_password'), 'rb') as fd:

        yield fd.read()

def test_http_url(sftpserver):

    with pytest.raises(ValueError):

        SFTP('https://coin.org/')

def test_missing_hostname(sftpserver):

    with pytest.raises(ValueError):

        SFTP('sftp://a:x@/hop/')

def test_sftp_ok(sftpserver):

    with sftpserver.serve_content({'DILA': {'a.zip': 'a'}}):

        with SFTP('sftp://john:doe@{server.host}:{server.port}/DILA/'.format(server=sftpserver)).client() as sftp:

            assert sftp.listdir() == ['a.zip']

def test_sftp_bad_paths(sftpserver):

    with sftpserver.serve_content({'DILA': {'a.zip': 'a'}}):

        with SFTP('sftp://john:doe@{server.host}:{server.port}/DILA/'.format(server=sftpserver)).client() as sftp:

            with pytest.raises(ValueError):

                sftp.chdir('..')

            with pytest.raises(ValueError):

                sftp.chdir('/')

            with pytest.raises(ValueError):

                sftp.chdir('/coin')

def test_form_field(sftpserver, ssh_key, ssh_key_with_password):

    from django import forms

    class Form(forms.Form):

        sftp = SFTPFormField(label='sftp')

    with sftpserver.serve_content({'DILA': {'a.zip': 'a'}}):

        url = 'sftp://john:doe@{server.host}:{server.port}/DILA/'.format(server=sftpserver)

        form = Form(data={'sftp_0': 'http://coin.org'})

        assert not form.is_valid()

        assert 'Enter a valid URL.' in str(form.errors)

        form = Form(data={'sftp_0': url})

        assert form.is_valid()

        sftp = form.cleaned_data['sftp']

        assert isinstance(sftp, SFTP)

        assert sftp.url == url

        assert sftp.username == 'john'

        assert sftp.password == 'doe'

        assert sftp.hostname == sftpserver.host

        assert sftp.port == sftpserver.port

        assert sftp.path == 'DILA'

        assert not sftp.private_key

        with form.cleaned_data['sftp'].client() as sftp:

            assert sftp.listdir() == ['a.zip']

        form = Form(data={'sftp_0': url, 'sftp_2': ssh_key.decode('ascii')})

        assert form.is_valid()

        sftp = form.cleaned_data['sftp']

        assert isinstance(sftp, SFTP)

        assert sftp.url == url

        assert sftp.username == 'john'

        assert sftp.password == 'doe'

        assert sftp.hostname == sftpserver.host

        assert sftp.port == sftpserver.port

        assert sftp.path == 'DILA'

        assert sftp.private_key

        with form.cleaned_data['sftp'].client() as sftp:

            assert sftp.listdir() == ['a.zip']

        form = Form(

            data={'sftp_0': url},

            files={'sftp_1': SimpleUploadedFile('ssh_key', ssh_key, 'application/octet-stream')})

        assert form.is_valid()

        sftp = form.cleaned_data['sftp']

        assert isinstance(sftp, SFTP)

        assert sftp.url == url

        assert sftp.username == 'john'

        assert sftp.password == 'doe'

        assert sftp.hostname == sftpserver.host

        assert sftp.port == sftpserver.port

        assert sftp.path == 'DILA'

        assert sftp.private_key

        with form.cleaned_data['sftp'].client() as sftp:

            assert sftp.listdir() == ['a.zip']

        form = Form(data={'sftp_0': url, 'sftp_2': ssh_key_with_password.decode('ascii')})

        assert not form.is_valid()

        assert 'key invalid' in str(form.errors)

        form = Form(data={

            'sftp_0': url,

            'sftp_2': ssh_key_with_password.decode('ascii'),

            'sftp_3': 'coucou',

        })

        assert form.is_valid()

        with form.cleaned_data['sftp'].client() as sftp:

            assert sftp.listdir() == ['a.zip']

@pytest.fixture

def temp_model(db):

    from django.db import connection

    class Model(models.Model):

        sftp = SFTPField(blank=True)

        class Meta:

            app_label = 'test'

    with connection.schema_editor() as editor:

        editor.create_model(Model)

        yield Model

        editor.delete_model(Model)

def test_model_field(temp_model, ssh_key_with_password):

    instance = temp_model.objects.create()

    assert instance.sftp is None

    url = 'sftp://john:doe@example.com:45/a/b'

    sftp = SFTP(url=url, private_key_content=ssh_key_with_password, private_key_password='coucou')

    instance.sftp = sftp

    instance.save()

    instance = temp_model.objects.get()

    assert instance.sftp is not None

    assert instance.sftp.url == url

    assert instance.sftp.private_key is not None

    instance.sftp = None

    instance.save()

    instance = temp_model.objects.get()

    assert instance.sftp is None

    instance.delete()

    temp_model.objects.create(sftp=sftp)

    instance = temp_model.objects.get()

    assert instance.sftp is not None

    assert instance.sftp.url == url

    assert instance.sftp.private_key is not None

  pytest-sftpserver