#!/usr/bin/env python

#

# Copyright 2011-2015 Splunk, Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License"): you may

# not use this file except in compliance with the License. You may obtain

# a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

# License for the specific language governing permissions and limitations

# under the License.

from __future__ import absolute_import

import pytest

from tests import testlib

import unittest

import splunklib.client as client

from splunklib.client import AuthenticationError

from splunklib.client import Service

from splunklib.binding import HTTPError

class ServiceTestCase(testlib.SDKTestCase):

def test_autologin(self):

service = client.connect(autologin=True, **self.opts.kwargs)

self.service.restart(timeout=120)

reader = service.jobs.oneshot("search index=internal | head 1")

self.assertIsNotNone(reader)

def test_capabilities(self):

capabilities = self.service.capabilities

self.assertTrue(isinstance(capabilities, list))

self.assertTrue(all([isinstance(c, str) for c in capabilities]))

self.assertTrue('change_own_password' in capabilities) # This should always be there...

def test_info(self):

info = self.service.info

keys = ["build", "cpu_arch", "guid", "isFree", "isTrial", "licenseKeys",

"licenseSignature", "licenseState", "master_guid", "mode",

"os_build", "os_name", "os_version", "serverName", "version"]

for key in keys:

self.assertTrue(key in list(info.keys()))

def test_info_with_namespace(self):

# Make sure we're not accessing /servicesNS/admin/search/server/info

# instead of /services/server/info

# Backup the values, which are probably set to None

owner, app = self.service.namespace["owner"], self.service.namespace["app"]

self.service.namespace["owner"] = self.service.username

self.service.namespace["app"] = "search"

try:

self.assertEqual(self.service.info.licenseState, 'OK')

except HTTPError as he:

self.fail("Couldn't get the server info, probably got a 403! %s" % he.message)

self.service.namespace["owner"] = owner

self.service.namespace["app"] = app

def test_without_namespace(self):

service = client.connect(**self.opts.kwargs)

service.apps.list()

def test_app_namespace(self):

kwargs = self.opts.kwargs.copy()

kwargs.update({'app': "search", 'owner': None})

service_ns = client.connect(**kwargs)

service_ns.apps.list()

def test_owner_wildcard(self):

kwargs = self.opts.kwargs.copy()

kwargs.update({'app': "search", 'owner': "-"})

service_ns = client.connect(**kwargs)

service_ns.apps.list()

def test_default_app(self):

kwargs = self.opts.kwargs.copy()

kwargs.update({'app': None, 'owner': "admin"})

service_ns = client.connect(**kwargs)

service_ns.apps.list()

def test_app_wildcard(self):

kwargs = self.opts.kwargs.copy()

kwargs.update({'app': "-", 'owner': "admin"})

service_ns = client.connect(**kwargs)

service_ns.apps.list()

def test_user_namespace(self):

kwargs = self.opts.kwargs.copy()

kwargs.update({'app': "search", 'owner': "admin"})

service_ns = client.connect(**kwargs)

service_ns.apps.list()

def test_parse(self):

# At the moment the parse method returns the raw XML. At

# some point this will change and it will return a nice,

# objectified form of the results, but for now there's

# nothing to test but a good response code.

response = self.service.parse('search * abc="def" | dedup abc')

self.assertEqual(response.status, 200)

def test_parse_fail(self):

try:

self.service.parse("xyzzy")

self.fail('Parse on nonsense did not fail')

except HTTPError as e:

self.assertEqual(e.status, 400)

def test_restart(self):

service = client.connect(**self.opts.kwargs)

self.service.restart(timeout=300)

service.login() # Make sure we are awake

def test_read_outputs_with_type(self):

name = testlib.tmpname()

service = client.connect(**self.opts.kwargs)

service.post('data/outputs/tcp/syslog', name=name, type='tcp')

entity = client.Entity(service, 'data/outputs/tcp/syslog/' + name)

self.assertTrue('tcp', entity.content.type)

if service.restart_required:

self.restartSplunk()

service = client.connect(**self.opts.kwargs)

client.Entity(service, 'data/outputs/tcp/syslog/' + name).delete()

if service.restart_required:

self.restartSplunk()

def test_splunk_version(self):

service = client.connect(**self.opts.kwargs)

v = service.splunk_version

self.assertTrue(isinstance(v, tuple))

self.assertTrue(len(v) >= 2)

for p in v:

self.assertTrue(isinstance(p, int) and p >= 0)

for version in [(4, 3, 3), (5,), (5, 0, 1)]:

with self.fake_splunk_version(version):

self.assertEqual(version, self.service.splunk_version)

def test_query_without_login_raises_auth_error(self):

service = self._create_unauthenticated_service()

self.assertRaises(AuthenticationError, lambda: service.indexes.list())

# This behavior is needed for backward compatibility for code

# prior to the introduction of AuthenticationError

def test_query_without_login_raises_http_401(self):

service = self._create_unauthenticated_service()

try:

service.indexes.list()

self.fail('Expected HTTP 401.')

except HTTPError as he:

if he.status == 401:

# Good

pass

else:

raise

def _create_unauthenticated_service(self):

return Service(**{

'host': self.opts.kwargs['host'],

'port': self.opts.kwargs['port'],

'scheme': self.opts.kwargs['scheme']

})

# To check the HEC event endpoint using Endpoint instance

@pytest.mark.smoke

def test_hec_event(self):

import json

service_hec = client.connect(host='localhost', scheme='https', port=8088,

token="11111111-1111-1111-1111-1111111111113")

event_collector_endpoint = client.Endpoint(service_hec, "/services/collector/event")

msg = {"index": "main", "event": "Hello World"}

response = event_collector_endpoint.post("", body=json.dumps(msg))

self.assertEqual(response.status, 200)

class TestCookieAuthentication(unittest.TestCase):

def setUp(self):

self.opts = testlib.parse([], {}, ".env")

self.service = client.Service(**self.opts.kwargs)

if getattr(unittest.TestCase, 'assertIsNotNone', None) is None:

def assertIsNotNone(self, obj, msg=None):

if obj is None:

raise self.failureException(msg or '%r is not None' % obj)

def test_login_and_store_cookie(self):

self.assertIsNotNone(self.service.get_cookies())

self.assertEqual(len(self.service.get_cookies()), 0)

self.service.login()

self.assertIsNotNone(self.service.get_cookies())

self.assertNotEqual(self.service.get_cookies(), {})

self.assertEqual(len(self.service.get_cookies()), 1)

def test_login_with_cookie(self):

self.service.login()

self.assertIsNotNone(self.service.get_cookies())

# Use the cookie from the other service as the only auth param (don't need user/password)

service2 = client.Service(**{"cookie": "%s=%s" % list(self.service.get_cookies().items())[0]})

service2.login()

self.assertEqual(len(service2.get_cookies()), 1)

self.assertEqual(service2.get_cookies(), self.service.get_cookies())

self.assertEqual(len(service2.get_cookies()), len(self.service.get_cookies()))

self.assertEqual(list(service2.get_cookies().keys())[0][:8], "splunkd_")

self.assertEqual(service2.apps.get().status, 200)

def test_login_fails_with_bad_cookie(self):

bad_cookie = {'bad': 'cookie'}

service2 = client.Service()

self.assertEqual(len(service2.get_cookies()), 0)

service2.get_cookies().update(bad_cookie)

self.assertEqual(service2.get_cookies(), {'bad': 'cookie'})

# Should get an error with a bad cookie

try:

service2.login()

self.fail()

except AuthenticationError as ae:

self.assertEqual(str(ae), "Login failed.")

def test_autologin_with_cookie(self):

self.service.login()

self.assertTrue(self.service.has_cookies())

service = client.connect(

autologin=True,

cookie="%s=%s" % list(self.service.get_cookies().items())[0],

**self.opts.kwargs)

self.assertTrue(service.has_cookies())

self.service.restart(timeout=120)

reader = service.jobs.oneshot("search index=internal | head 1")

self.assertIsNotNone(reader)

def test_login_fails_with_no_cookie(self):

service2 = client.Service()

self.assertEqual(len(service2.get_cookies()), 0)

# Should get an error when no authentication method

try:

service2.login()

self.fail()

except AuthenticationError as ae:

self.assertEqual(str(ae), "Login failed.")

def test_login_with_multiple_cookie_headers(self):

cookies = {

'bad': 'cookie',

'something_else': 'bad'

}

self.service.logout()

self.service.get_cookies().update(cookies)

self.service.login()

self.assertEqual(self.service.apps.get().status, 200)

def test_login_with_multiple_cookies(self):

bad_cookie = 'bad=cookie'

self.service.login()

self.assertIsNotNone(self.service.get_cookies())

service2 = client.Service(**{"cookie": bad_cookie})

# Should get an error with a bad cookie

try:

service2.login()

self.fail()

except AuthenticationError as ae:

self.assertEqual(str(ae), "Login failed.")

# Add on valid cookies, and try to use all of them

service2.get_cookies().update(self.service.get_cookies())

self.assertEqual(len(service2.get_cookies()), 2)

self.service.get_cookies().update({'bad': 'cookie'})

self.assertEqual(service2.get_cookies(), self.service.get_cookies())

self.assertEqual(len(service2.get_cookies()), 2)

self.assertTrue([cookie for cookie in service2.get_cookies() if "splunkd_" in cookie])

self.assertTrue('bad' in service2.get_cookies())

self.assertEqual(service2.get_cookies()['bad'], 'cookie')

self.assertEqual(set(self.service.get_cookies()), set(service2.get_cookies()))

service2.login()

self.assertEqual(service2.apps.get().status, 200)

class TestSettings(testlib.SDKTestCase):

def test_read_settings(self):

settings = self.service.settings

# Verify that settings contains the keys we expect

keys = [

"SPLUNK_DB", "SPLUNK_HOME", "enableSplunkWebSSL", "host",

"httpport", "mgmtHostPort", "minFreeSpace", "pass4SymmKey",

"serverName", "sessionTimeout", "startwebserver", "trustedIP"

]

for key in keys:

self.assertTrue(key in settings)

def test_update_settings(self):

settings = self.service.settings

# Verify that we can update the settings

original = settings['sessionTimeout']

self.assertTrue(original != "42h")

settings.update(sessionTimeout="42h")

settings.refresh()

updated = settings['sessionTimeout']

self.assertEqual(updated, "42h")

# Restore (and verify) original value

settings.update(sessionTimeout=original)

settings.refresh()

updated = settings['sessionTimeout']

self.assertEqual(updated, original)

self.restartSplunk()

class TestTrailing(unittest.TestCase):

template = '/servicesNS/boris/search/another/path/segment/that runs on'

def test_raises_when_not_found_first(self):

self.assertRaises(ValueError, client._trailing, 'this is a test', 'boris')

def test_raises_when_not_found_second(self):

self.assertRaises(ValueError, client._trailing, 'this is a test', 's is', 'boris')

def test_no_args_is_identity(self):

self.assertEqual(self.template, client._trailing(self.template))

def test_trailing_with_one_arg_works(self):

self.assertEqual('boris/search/another/path/segment/that runs on',

client._trailing(self.template, 'ervicesNS/'))

def test_trailing_with_n_args_works(self):

self.assertEqual(

'another/path/segment/that runs on',

client._trailing(self.template, 'servicesNS/', '/', '/')

)

class TestEntityNamespacing(testlib.SDKTestCase):

def test_proper_namespace_with_arguments(self):

entity = self.service.apps['search']

self.assertEqual((None, None, "global"), entity._proper_namespace(sharing="global"))

self.assertEqual((None, "search", "app"), entity._proper_namespace(sharing="app", app="search"))

self.assertEqual(

("admin", "search", "user"),

entity._proper_namespace(sharing="user", app="search", owner="admin")

)

def test_proper_namespace_with_entity_namespace(self):

entity = self.service.apps['search']

namespace = (entity.access.owner, entity.access.app, entity.access.sharing)

self.assertEqual(namespace, entity._proper_namespace())

def test_proper_namespace_with_service_namespace(self):

entity = client.Entity(self.service, client.PATH_APPS + "search")

del entity._state['access']

namespace = (self.service.namespace.owner,

self.service.namespace.app,

self.service.namespace.sharing)

self.assertEqual(namespace, entity._proper_namespace())

if __name__ == "__main__":

try:

import unittest2 as unittest

except ImportError:

import unittest

unittest.main()