salt/tests/integration/utils/test_smb.py

# -*- coding: utf-8 -*-
"""
Test utility methods that communicate with SMB shares.
"""
from __future__ import absolute_import

import getpass
import logging
import os
import signal
import subprocess
import tempfile
import time

import pytest
import salt.utils.files
import salt.utils.path
import salt.utils.smb
from tests.support.case import TestCase

log = logging.getLogger(__name__)
CONFIG = (
    "[global]\n"
    "realm = saltstack.com\n"
    "interfaces = lo 127.0.0.0/8\n"
    "smb ports = 1445\n"
    "log level = 2\n"
    "map to guest = Bad User\n"
    "enable core files = no\n"
    "passdb backend = smbpasswd\n"
    "smb passwd file = {passwdb}\n"
    "lock directory = {samba_dir}\n"
    "state directory = {samba_dir}\n"
    "cache directory = {samba_dir}\n"
    "pid directory = {samba_dir}\n"
    "private dir = {samba_dir}\n"
    "ncalrpc dir = {samba_dir}\n"
    "socket options = IPTOS_LOWDELAY TCP_NODELAY\n"
    "min receivefile size = 0\n"
    "write cache size = 0\n"
    "client ntlmv2 auth = no\n"
    "client min protocol = SMB3_11\n"
    "client plaintext auth = no\n"
    "\n"
    "[public]\n"
    "path = {public_dir}\n"
    "read only = no\n"
    "guest ok = no\n"
    "writeable = yes\n"
    "force user = {user}\n"
)
TBE = (
    "{}:0:XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX:AC8E657F8"
    "3DF82BEEA5D43BDAF7800CC:[U          ]:LCT-507C14C7:"
)


def which_smbd():
    """
    Find the smbd executable and cache the result if it exits.
    """
    if hasattr(which_smbd, "cached_result"):
        return which_smbd.cached_result
    smbd = salt.utils.path.which("smbd")
    if smbd:
        which_smbd.cached_result = smbd
    return smbd


@pytest.mark.skipif(not which_smbd(), reason="smbd binary not found")
@pytest.mark.skipif(
    any([salt.utils.smb.HAS_IMPACKET, salt.utils.smb.HAS_SMBPROTOCOL]),
    reason='Either "impacket" or "smbprotocol" needs to be installed.',
)
class TestSmb(TestCase):

    _smbd = None

    @staticmethod
    def check_pid(pid):
        try:
            os.kill(pid, 0)
        except OSError:
            return False
        else:
            return True

    @classmethod
    def setUpClass(cls):
        tmpdir = tempfile.mkdtemp()
        cls.samba_dir = os.path.join(tmpdir, "samba")
        cls.public_dir = os.path.join(tmpdir, "public")
        os.makedirs(cls.samba_dir)
        os.makedirs(cls.public_dir)
        os.chmod(cls.samba_dir, 0o775)
        os.chmod(cls.public_dir, 0o775)
        passwdb = os.path.join(tmpdir, "passwdb")
        cls.username = getpass.getuser()
        with salt.utils.files.fopen(passwdb, "w") as fp:
            fp.write(TBE.format(cls.username))
        samba_conf = os.path.join(tmpdir, "smb.conf")
        with salt.utils.files.fopen(samba_conf, "w") as fp:
            fp.write(
                CONFIG.format(
                    samba_dir=cls.samba_dir,
                    public_dir=cls.public_dir,
                    passwdb=passwdb,
                    user=cls.username,
                )
            )
        cls._smbd = subprocess.Popen(
            "{0} -FS -P0 -s {1}".format(which_smbd(), samba_conf), shell=True
        )
        time.sleep(1)
        pidfile = os.path.join(cls.samba_dir, "smbd.pid")
        with salt.utils.files.fopen(pidfile, "r") as fp:
            cls._pid = int(fp.read().strip())
        if not cls.check_pid(cls._pid):
            raise Exception("Unable to locate smbd's pid file")

    @classmethod
    def tearDownClass(cls):
        log.warning("teardown")
        os.kill(cls._pid, signal.SIGTERM)

    def test_write_file(self):
        """
        Transfer a file over SMB
        """
        name = "test_write_file.txt"
        content = "write test file content"
        share_path = os.path.join(self.public_dir, name)
        assert not os.path.exists(share_path)

        local_path = tempfile.mktemp()
        with salt.utils.files.fopen(local_path, "w") as fp:
            fp.write(content)
        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        salt.utils.smb.put_file(local_path, name, "public", conn=conn)
        conn.close()

        assert os.path.exists(share_path)
        with salt.utils.files.fopen(share_path, "r") as fp:
            result = fp.read()
        assert result == content

    def test_write_str(self):
        """
        Write a string to a file over SMB
        """
        name = "test_write_str.txt"
        content = "write test file content"
        share_path = os.path.join(self.public_dir, name)
        assert not os.path.exists(share_path)
        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        salt.utils.smb.put_str(content, name, "public", conn=conn)
        conn.close()

        assert os.path.exists(share_path)
        with salt.utils.files.fopen(share_path, "r") as fp:
            result = fp.read()
        assert result == content

    def test_delete_file(self):
        """
        Validate deletion of files over SMB
        """
        name = "test_delete_file.txt"
        content = "read test file content"
        share_path = os.path.join(self.public_dir, name)
        with salt.utils.files.fopen(share_path, "w") as fp:
            fp.write(content)
        assert os.path.exists(share_path)

        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        salt.utils.smb.delete_file(name, "public", conn=conn)
        conn.close()

        assert not os.path.exists(share_path)

    def test_mkdirs(self):
        """
        Create directories over SMB
        """
        dir_name = "mkdirs/test"
        share_path = os.path.join(self.public_dir, dir_name)
        assert not os.path.exists(share_path)

        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        salt.utils.smb.mkdirs(dir_name, "public", conn=conn)
        conn.close()

        assert os.path.exists(share_path)

    def test_delete_dirs(self):
        """
        Validate deletion of directoreies over SMB
        """
        dir_name = "deldirs"
        subdir_name = "deldirs/test"
        local_path = os.path.join(self.public_dir, subdir_name)
        os.makedirs(local_path)
        assert os.path.exists(local_path)

        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        salt.utils.smb.delete_directory(subdir_name, "public", conn=conn)
        conn.close()

        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        salt.utils.smb.delete_directory(dir_name, "public", conn=conn)
        conn.close()

        assert not os.path.exists(local_path)
        assert not os.path.exists(os.path.join(self.public_dir, dir_name))

    def test_connection(self):
        """
        Validate creation of an SMB connection
        """
        conn = salt.utils.smb.get_conn("127.0.0.1", self.username, "foo", port=1445)
        conn.close()