python/samba/tests/dcerpc/dnsserver.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Amitay Isaacs <amitay@gmail.com> 2011
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 """Tests for samba.dcerpc.dnsserver"""
  19
  20 from samba.dcerpc import dnsp, dnsserver
  21 from samba.tests import RpcInterfaceTestCase, env_get_var_value
  22 from samba.netcmd.dns import ARecord
  23
  24 class DnsserverTests(RpcInterfaceTestCase):
  25
  26     def setUp(self):
  27         super(DnsserverTests, self).setUp()
  28         self.server = env_get_var_value("SERVER_IP")
  29         self.zone = env_get_var_value("REALM").lower()
  30         self.conn = dnsserver.dnsserver("ncacn_ip_tcp:%s" % (self.server),
  31                                         self.get_loadparm(),
  32                                         self.get_credentials())
  33
  34     def test_operation2(self):
  35         pass
  36
  37
  38     def test_query2(self):
  39         typeid, result = self.conn.DnssrvQuery2(dnsserver.DNS_CLIENT_VERSION_W2K,
  40                                                 0,
  41                                                 self.server,
  42                                                 None,
  43                                                 'ServerInfo')
  44         self.assertEquals(dnsserver.DNSSRV_TYPEID_SERVER_INFO_W2K, typeid)
  45
  46         typeid, result = self.conn.DnssrvQuery2(dnsserver.DNS_CLIENT_VERSION_DOTNET,
  47                                                 0,
  48                                                 self.server,
  49                                                 None,
  50                                                 'ServerInfo')
  51         self.assertEquals(dnsserver.DNSSRV_TYPEID_SERVER_INFO_DOTNET, typeid)
  52
  53         typeid, result = self.conn.DnssrvQuery2(dnsserver.DNS_CLIENT_VERSION_LONGHORN,
  54                                                 0,
  55                                                 self.server,
  56                                                 None,
  57                                                 'ServerInfo')
  58         self.assertEquals(dnsserver.DNSSRV_TYPEID_SERVER_INFO, typeid)
  59
  60     def test_operation2(self):
  61         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
  62         rev_zone = '1.168.192.in-addr.arpa'
  63
  64         zone_create = dnsserver.DNS_RPC_ZONE_CREATE_INFO_LONGHORN()
  65         zone_create.pszZoneName = rev_zone
  66         zone_create.dwZoneType = dnsp.DNS_ZONE_TYPE_PRIMARY
  67         zone_create.fAllowUpdate = dnsp.DNS_ZONE_UPDATE_SECURE
  68         zone_create.fAging = 0
  69         zone_create.dwDpFlags = dnsserver.DNS_DP_DOMAIN_DEFAULT
  70
  71         # Create zone
  72         self.conn.DnssrvOperation2(client_version,
  73                                     0,
  74                                     self.server,
  75                                     None,
  76                                     0,
  77                                     'ZoneCreate',
  78                                     dnsserver.DNSSRV_TYPEID_ZONE_CREATE,
  79                                     zone_create)
  80
  81         request_filter = (dnsserver.DNS_ZONE_REQUEST_REVERSE |
  82                             dnsserver.DNS_ZONE_REQUEST_PRIMARY)
  83         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
  84                                                             0,
  85                                                             self.server,
  86                                                             None,
  87                                                             'EnumZones',
  88                                                             dnsserver.DNSSRV_TYPEID_DWORD,
  89                                                             request_filter)
  90         self.assertEquals(1, zones.dwZoneCount)
  91
  92         # Delete zone
  93         self.conn.DnssrvOperation2(client_version,
  94                                     0,
  95                                     self.server,
  96                                     rev_zone,
  97                                     0,
  98                                     'DeleteZoneFromDs',
  99                                     dnsserver.DNSSRV_TYPEID_NULL,
 100                                     None)
 101
 102         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
 103                                                             0,
 104                                                             self.server,
 105                                                             None,
 106                                                             'EnumZones',
 107                                                             dnsserver.DNSSRV_TYPEID_DWORD,
 108                                                             request_filter)
 109         self.assertEquals(0, zones.dwZoneCount)
 110
 111
 112     def test_complexoperation2(self):
 113         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
 114         request_filter = (dnsserver.DNS_ZONE_REQUEST_FORWARD |
 115                             dnsserver.DNS_ZONE_REQUEST_PRIMARY)
 116         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
 117                                                             0,
 118                                                             self.server,
 119                                                             None,
 120                                                             'EnumZones',
 121                                                             dnsserver.DNSSRV_TYPEID_DWORD,
 122                                                             request_filter)
 123         self.assertEquals(dnsserver.DNSSRV_TYPEID_ZONE_LIST, typeid)
 124         self.assertEquals(2, zones.dwZoneCount)
 125
 126         request_filter = (dnsserver.DNS_ZONE_REQUEST_REVERSE |
 127                             dnsserver.DNS_ZONE_REQUEST_PRIMARY)
 128         typeid, zones = self.conn.DnssrvComplexOperation2(client_version,
 129                                                             0,
 130                                                             self.server,
 131                                                             None,
 132                                                             'EnumZones',
 133                                                             dnsserver.DNSSRV_TYPEID_DWORD,
 134                                                             request_filter)
 135         self.assertEquals(dnsserver.DNSSRV_TYPEID_ZONE_LIST, typeid)
 136         self.assertEquals(0, zones.dwZoneCount)
 137
 138
 139     def test_enumrecords2(self):
 140         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
 141         record_type = dnsp.DNS_TYPE_NS
 142         select_flags = (dnsserver.DNS_RPC_VIEW_ROOT_HINT_DATA |
 143                         dnsserver.DNS_RPC_VIEW_ADDITIONAL_DATA)
 144         buflen, roothints = self.conn.DnssrvEnumRecords2(client_version,
 145                                                             0,
 146                                                             self.server,
 147                                                             '..RootHints',
 148                                                             '.',
 149                                                             None,
 150                                                             record_type,
 151                                                             select_flags,
 152                                                             None,
 153                                                             None)
 154         self.assertEquals(14, roothints.count)  # 1 NS + 13 A records (a-m)
 155
 156
 157     def test_updaterecords2(self):
 158         client_version = dnsserver.DNS_CLIENT_VERSION_LONGHORN
 159         record_type = dnsp.DNS_TYPE_A
 160         select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA
 161
 162         name = 'dummy'
 163         rec = ARecord('1.2.3.4')
 164         rec2 = ARecord('5.6.7.8')
 165
 166         # Add record
 167         add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
 168         add_rec_buf.rec = rec
 169         self.conn.DnssrvUpdateRecord2(client_version,
 170                                         0,
 171                                         self.server,
 172                                         self.zone,
 173                                         name,
 174                                         add_rec_buf,
 175                                         None)
 176
 177         buflen, result = self.conn.DnssrvEnumRecords2(client_version,
 178                                                         0,
 179                                                         self.server,
 180                                                         self.zone,
 181                                                         name,
 182                                                         None,
 183                                                         record_type,
 184                                                         select_flags,
 185                                                         None,
 186                                                         None)
 187         self.assertEquals(1, result.count)
 188         self.assertEquals(1, result.rec[0].wRecordCount)
 189         self.assertEquals(dnsp.DNS_TYPE_A, result.rec[0].records[0].wType)
 190         self.assertEquals('1.2.3.4', result.rec[0].records[0].data)
 191
 192         # Update record
 193         add_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
 194         add_rec_buf.rec = rec2
 195         del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
 196         del_rec_buf.rec = rec
 197         self.conn.DnssrvUpdateRecord2(client_version,
 198                                         0,
 199                                         self.server,
 200                                         self.zone,
 201                                         name,
 202                                         add_rec_buf,
 203                                         del_rec_buf)
 204
 205         buflen, result = self.conn.DnssrvEnumRecords2(client_version,
 206                                                         0,
 207                                                         self.server,
 208                                                         self.zone,
 209                                                         name,
 210                                                         None,
 211                                                         record_type,
 212                                                         select_flags,
 213                                                         None,
 214                                                         None)
 215         self.assertEquals(1, result.count)
 216         self.assertEquals(1, result.rec[0].wRecordCount)
 217         self.assertEquals(dnsp.DNS_TYPE_A, result.rec[0].records[0].wType)
 218         self.assertEquals('5.6.7.8', result.rec[0].records[0].data)
 219
 220         # Delete record
 221         del_rec_buf = dnsserver.DNS_RPC_RECORD_BUF()
 222         del_rec_buf.rec = rec2
 223         self.conn.DnssrvUpdateRecord2(client_version,
 224                                         0,
 225                                         self.server,
 226                                         self.zone,
 227                                         name,
 228                                         None,
 229                                         del_rec_buf)
 230
 231         self.assertRaises(RuntimeError, self.conn.DnssrvEnumRecords2,
 232                                         client_version,
 233                                         0,
 234                                         self.server,
 235                                         self.zone,
 236                                         name,
 237                                         None,
 238                                         record_type,
 239                                         select_flags,
 240                                         None,
 241                                         None)