mirror of
https://github.com/cuckoosandbox/cuckoo.git
synced 2026-10-12 02:12:40 +00:00
331 lines
9.1 KiB
Python
331 lines
9.1 KiB
Python
# Copyright (C) 2016-2018 Cuckoo Foundation.
|
|
# This file is part of Cuckoo Sandbox - http://www.cuckoosandbox.org
|
|
# See the file 'docs/LICENSE' for copying permission.
|
|
|
|
import ctypes
|
|
import mock
|
|
import os
|
|
import pytest
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
from cuckoo.common.exceptions import CuckooStartupError
|
|
from cuckoo.common.files import Files
|
|
from cuckoo.common.structures import Structure
|
|
from cuckoo.main import cuckoo_create
|
|
from cuckoo.misc import (
|
|
dispatch, cwd, set_cwd, getuser, mkdir, Popen, drop_privileges, make_list,
|
|
HAVE_PWD, is_linux, is_windows, is_macosx, decide_cwd, Pidfile,
|
|
format_command
|
|
)
|
|
|
|
def return_value(value):
|
|
return value
|
|
|
|
def sleep2(value):
|
|
time.sleep(2)
|
|
return value
|
|
|
|
def test_dispatch():
|
|
assert dispatch(return_value, (1,)) == 1
|
|
assert dispatch(return_value, ("foo",)) == "foo"
|
|
|
|
assert dispatch(sleep2, (2,)) == 2
|
|
assert dispatch(sleep2, (2,), timeout=1) is None
|
|
|
|
with pytest.raises(RuntimeError):
|
|
dispatch(None, args=None)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
dispatch(None, kwargs=None)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
dispatch(None, process=False)
|
|
|
|
def test_cwd():
|
|
set_cwd("/tmp/foo")
|
|
assert cwd() == "/tmp/foo"
|
|
assert cwd("a") == os.path.join("/tmp/foo", "a")
|
|
assert cwd("a", "b") == os.path.join("/tmp/foo", "a", "b")
|
|
|
|
set_cwd("/home/user/.cuckoo", "~/.cuckoo")
|
|
assert cwd(raw=True) == "~/.cuckoo"
|
|
assert cwd(root=True) == "/home/user/.cuckoo"
|
|
assert cwd("dump.pcap", analysis=1234) == os.path.join(
|
|
"/home/user/.cuckoo", "storage", "analyses", "1234", "dump.pcap"
|
|
)
|
|
|
|
assert os.path.exists(cwd("guids.txt", private=True))
|
|
|
|
with pytest.raises(RuntimeError):
|
|
cwd("foo", private=False)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
cwd("foo", raw=False)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
cwd("foo", root=False)
|
|
|
|
with pytest.raises(RuntimeError):
|
|
cwd("foo", analysis=None)
|
|
|
|
@pytest.mark.skipif(not HAVE_PWD, reason="This test is not for Windows")
|
|
def test_getuser():
|
|
# TODO This probably doesn't work on all platforms.
|
|
assert getuser() == subprocess.check_output(["id", "-un"]).strip()
|
|
|
|
def test_mkdir():
|
|
dirpath = tempfile.mkdtemp()
|
|
assert os.path.isdir(dirpath)
|
|
mkdir(dirpath)
|
|
assert os.path.isdir(dirpath)
|
|
|
|
dirpath = tempfile.mktemp()
|
|
assert not os.path.exists(dirpath)
|
|
mkdir(dirpath)
|
|
assert os.path.isdir(dirpath)
|
|
|
|
@pytest.mark.skipif("sys.platform != 'win32'")
|
|
def test_is_windows():
|
|
assert is_windows() is True
|
|
assert is_linux() is False
|
|
assert is_macosx() is False
|
|
|
|
@pytest.mark.skipif("sys.platform != 'darwin'")
|
|
def test_is_macosx():
|
|
assert is_windows() is False
|
|
assert is_linux() is False
|
|
assert is_macosx() is True
|
|
|
|
@pytest.mark.skipif("sys.platform != 'linux2'")
|
|
def test_is_linux():
|
|
assert is_windows() is False
|
|
assert is_linux() is True
|
|
assert is_macosx() is False
|
|
|
|
def test_platforms():
|
|
"""Ensure that the above unit tests are complete (for our supported
|
|
platforms)."""
|
|
assert sys.platform in ("win32", "linux2", "darwin")
|
|
|
|
def test_popen():
|
|
"""Ensure that Popen is working properly."""
|
|
with mock.patch("subprocess.Popen") as p:
|
|
p.return_value = None
|
|
Popen(["foo", "bar"])
|
|
|
|
p.assert_called_once_with(["foo", "bar"])
|
|
|
|
with mock.patch("subprocess.Popen") as p:
|
|
p.return_value = None
|
|
Popen(
|
|
["foo", "bar"], close_fds=True,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
|
)
|
|
|
|
if is_windows():
|
|
p.assert_called_once_with(
|
|
["foo", "bar"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
|
)
|
|
else:
|
|
p.assert_called_once_with(
|
|
["foo", "bar"], close_fds=True,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
|
)
|
|
|
|
# Test that the method actually works.
|
|
p = Popen("echo 123", stdout=subprocess.PIPE, shell=True)
|
|
out, err = p.communicate()
|
|
assert out.strip() == "123" and not err
|
|
|
|
# The following would normally throw an exception on Windows.
|
|
p = Popen("echo 1234", close_fds=True, stdout=subprocess.PIPE, shell=True)
|
|
out, err = p.communicate()
|
|
assert out.strip() == "1234" and not err
|
|
|
|
def test_decide_cwd():
|
|
orig_cuckoo_cwd = os.environ.pop("CUCKOO_CWD", None)
|
|
orig_cuckoo = os.environ.pop("CUCKOO", None)
|
|
|
|
dirpath1 = tempfile.mkdtemp()
|
|
dirpath2 = tempfile.mkdtemp()
|
|
dirpath3 = tempfile.mkdtemp()
|
|
|
|
assert decide_cwd(dirpath1) == dirpath1
|
|
|
|
assert decide_cwd() == os.path.abspath(os.path.expanduser("~/.cuckoo"))
|
|
|
|
curdir = os.getcwd()
|
|
os.chdir(dirpath2)
|
|
open(".cwd", "wb").write("A"*40)
|
|
|
|
assert decide_cwd() == os.path.abspath(".")
|
|
os.chdir(curdir)
|
|
|
|
os.environ["CUCKOO"] = dirpath2
|
|
assert decide_cwd(dirpath1) == dirpath1
|
|
assert decide_cwd() == dirpath2
|
|
|
|
os.environ["CUCKOO_CWD"] = dirpath3
|
|
assert decide_cwd(dirpath1) == dirpath1
|
|
assert decide_cwd() == dirpath3
|
|
|
|
with pytest.raises(CuckooStartupError) as e:
|
|
decide_cwd(tempfile.mktemp(), exists=True)
|
|
e.match("is not present")
|
|
|
|
with pytest.raises(CuckooStartupError) as e:
|
|
decide_cwd(dirpath1, exists=True)
|
|
e.match("is not a proper CWD")
|
|
|
|
Files.create(dirpath1, ".cwd", "A"*40)
|
|
assert decide_cwd(dirpath1, exists=True) == dirpath1
|
|
|
|
# Cleanup.
|
|
if orig_cuckoo:
|
|
os.environ["CUCKOO"] = orig_cuckoo
|
|
else:
|
|
os.environ.pop("CUCKOO", None)
|
|
|
|
if orig_cuckoo_cwd:
|
|
os.environ["CUCKOO_CWD"] = orig_cuckoo_cwd
|
|
else:
|
|
os.environ.pop("CUCKOO_CWD", None)
|
|
|
|
@pytest.mark.skipif("sys.platform != 'linux2'")
|
|
@mock.patch("cuckoo.misc.pwd")
|
|
@mock.patch("cuckoo.misc.os")
|
|
def test_drop_privileges(p, q):
|
|
drop_privileges("username")
|
|
q.getpwnam.assert_called_once_with("username")
|
|
p.setgroups.assert_called_once()
|
|
p.setgid.assert_called_once()
|
|
p.setuid.assert_called_once()
|
|
p.putenv.assert_called_once()
|
|
|
|
def test_structure():
|
|
class S1(Structure):
|
|
_pack_ = 1
|
|
_fields_ = [
|
|
("a", ctypes.c_ubyte),
|
|
("b", ctypes.c_ushort),
|
|
("c", ctypes.c_uint),
|
|
("d", ctypes.c_ubyte * 128),
|
|
]
|
|
|
|
class S2(Structure):
|
|
_pack_ = 1
|
|
_fields_ = [
|
|
("a", S1),
|
|
("b", ctypes.c_ulonglong),
|
|
("c", ctypes.c_char * 32),
|
|
]
|
|
|
|
a = S2.from_buffer_copy("A"*175)
|
|
assert a.a.a == 0x41
|
|
assert a.a.b == 0x4141
|
|
assert a.a.c == 0x41414141
|
|
assert a.a.d[:] == [0x41] * 128
|
|
assert a.b == 0x4141414141414141
|
|
assert a.c == "A"*32
|
|
assert a.as_dict() == {
|
|
"a": {
|
|
"a": 0x41,
|
|
"b": 0x4141,
|
|
"c": 0x41414141,
|
|
"d": [0x41] * 128,
|
|
},
|
|
"b": 0x4141414141414141,
|
|
"c": "A"*32,
|
|
}
|
|
|
|
def test_create_pidfile():
|
|
set_cwd(tempfile.mkdtemp())
|
|
cuckoo_create()
|
|
|
|
Pidfile("test1").create()
|
|
assert int(open(cwd("pidfiles", "test1.pid"), "rb").read()) == os.getpid()
|
|
|
|
def test_remove_pidfile():
|
|
set_cwd(tempfile.mkdtemp())
|
|
cuckoo_create()
|
|
|
|
Pidfile("test2").create()
|
|
assert os.path.exists(cwd("pidfiles", "test2.pid"))
|
|
|
|
Pidfile("test2").remove()
|
|
assert not os.path.exists(cwd("pidfiles", "test2.pid"))
|
|
|
|
def test_pidfile_exists_false():
|
|
set_cwd(tempfile.mkdtemp())
|
|
cuckoo_create()
|
|
assert not Pidfile("test3").exists()
|
|
|
|
def test_pidfile_exists_true():
|
|
set_cwd(tempfile.mkdtemp())
|
|
cuckoo_create()
|
|
|
|
p = Pidfile("test4")
|
|
p.create()
|
|
assert p.exists() and p.pid == os.getpid()
|
|
|
|
def test_pidfile_none():
|
|
set_cwd(tempfile.mkdtemp())
|
|
cuckoo_create()
|
|
|
|
p = Pidfile("test5")
|
|
p.create()
|
|
|
|
open(cwd("pidfiles", "test5.pid"), "wb").write("notapid")
|
|
assert p.read() is None
|
|
|
|
def test_proc_exists():
|
|
assert Pidfile("hello").proc_exists(os.getpid())
|
|
# Chances are possible, but slim.
|
|
assert not Pidfile("hello").proc_exists(13337)
|
|
|
|
assert not Pidfile("hello)").proc_exists(None)
|
|
assert not Pidfile("hello)").proc_exists("")
|
|
|
|
@mock.patch("cuckoo.misc.sys")
|
|
def test_pid_exists_unsupported_platform(p):
|
|
p.platform = "DogeOS"
|
|
assert Pidfile("hello").proc_exists(os.getpid()) is None
|
|
|
|
def test_active_pids():
|
|
set_cwd(tempfile.mkdtemp())
|
|
cuckoo_create()
|
|
|
|
Pidfile("test6").create()
|
|
assert Pidfile.get_active_pids() == {
|
|
"test6": os.getpid(),
|
|
}
|
|
|
|
def test_make_list():
|
|
assert make_list("hello") == ["hello"]
|
|
assert make_list(1) == [1]
|
|
assert make_list((1, 2)) == [1, 2]
|
|
assert make_list([3, 4]) == [3, 4]
|
|
|
|
def test_format_command():
|
|
set_cwd(tempfile.mkdtemp(), ".")
|
|
assert format_command("community") == "cuckoo community"
|
|
|
|
set_cwd(tempfile.mkdtemp(), "~/.cuckoo")
|
|
assert format_command("community") == "cuckoo community"
|
|
|
|
dirpath = tempfile.mkdtemp()
|
|
set_cwd(dirpath, dirpath)
|
|
assert format_command("community") == "cuckoo --cwd %s community" % cwd()
|
|
|
|
dirpath = tempfile.mkdtemp("foo bar")
|
|
set_cwd(dirpath, dirpath)
|
|
assert format_command("community") == 'cuckoo --cwd "%s" community' % cwd()
|
|
|
|
dirpath = tempfile.mkdtemp("foo ' bar")
|
|
set_cwd(dirpath, dirpath)
|
|
assert format_command("community") == 'cuckoo --cwd "%s" community' % cwd()
|