File: dds.py

package info (click to toggle)
python-ldap 3.1.0-2
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 2,248 kB
  • sloc: python: 9,465; ansic: 2,828; makefile: 132; sh: 68
file content (75 lines) | stat: -rw-r--r-- 2,093 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
# -*- coding: utf-8 -*-
"""
ldap.extop.dds - Classes for Dynamic Entries extended operations
(see RFC 2589)

See https://www.python-ldap.org/ for details.
"""

from ldap.extop import ExtendedRequest,ExtendedResponse

# Imports from pyasn1
from pyasn1.type import namedtype,univ,tag
from pyasn1.codec.der import encoder,decoder
from pyasn1_modules.rfc2251 import LDAPDN


class RefreshRequest(ExtendedRequest):

  requestName = '1.3.6.1.4.1.1466.101.119.1'
  defaultRequestTtl = 86400

  class RefreshRequestValue(univ.Sequence):
    componentType = namedtype.NamedTypes(
      namedtype.NamedType(
        'entryName',
        LDAPDN().subtype(
          implicitTag=tag.Tag(tag.tagClassContext,tag.tagFormatSimple,0)
        )
      ),
      namedtype.NamedType(
        'requestTtl',
        univ.Integer().subtype(
          implicitTag=tag.Tag(tag.tagClassContext,tag.tagFormatSimple,1)
        )
      ),
    )

  def __init__(self,requestName=None,entryName=None,requestTtl=None):
    self.entryName = entryName
    self.requestTtl = requestTtl or self.defaultRequestTtl

  def encodedRequestValue(self):
    p = self.RefreshRequestValue()
    p.setComponentByName(
      'entryName',
      LDAPDN(self.entryName).subtype(
        implicitTag=tag.Tag(tag.tagClassContext, tag.tagFormatSimple,0)
      )
    )
    p.setComponentByName(
      'requestTtl',
      univ.Integer(self.requestTtl).subtype(
        implicitTag=tag.Tag(tag.tagClassContext,tag.tagFormatSimple,1)
      )
    )
    return encoder.encode(p)


class RefreshResponse(ExtendedResponse):
  responseName = '1.3.6.1.4.1.1466.101.119.1'

  class RefreshResponseValue(univ.Sequence):
    componentType = namedtype.NamedTypes(
      namedtype.NamedType(
        'responseTtl',
        univ.Integer().subtype(
          implicitTag=tag.Tag(tag.tagClassContext,tag.tagFormatSimple,1)
        )
      )
    )

  def decodeResponseValue(self,value):
    respValue,_ = decoder.decode(value,asn1Spec=self.RefreshResponseValue())
    self.responseTtl = int(respValue.getComponentByName('responseTtl'))
    return self.responseTtl