1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 """helpers for dealing with Twisted PB connections.
19 """
20
21 import re
22
23 from twisted.spread import pb
24
25 from flumotion.twisted import pb as fpb
26
27 __version__ = "$Rev$"
28
29
31 """
32 I hold information on how to connect to a PB server somewhere. I can
33 be transferred over the wire.
34 """
35
36 - def __init__(self, host, port, use_ssl, authenticator):
37 self.host = host
38 self.port = port
39 self.use_ssl = use_ssl
40 self.authenticator = authenticator
41
43
44
45
46 if (self.authenticator
47 and getattr(self.authenticator, 'username', None)):
48 return '%s@%s:%d' % (self.authenticator.username,
49 self.host, self.port)
50 else:
51 return '%s:%d' % (self.host, self.port)
52
53 pb.setUnjellyableForClass(PBConnectionInfo, PBConnectionInfo)
54
55
56 _pat = re.compile('^(([^:@]*)(:([^:@]+))?@)?([^:@]+)(:([0-9]+))?$')
57
58
61 """
62 Parse a string representation of a PB connection into a
63 PBConnectionInfo object.
64
65 The expected format is [user[:pass]@]host[:port]. Only the host is
66 mandatory. The default values for username, password, and port will
67 be taken from the optional username, password and port arguments.
68
69 @param string: A string describing the PB connection.
70 @type string: str
71 @param username: Default username, or 'user' if not given.
72 @type username: str
73 @param password: Default password, or 'test' if not given.
74 @type password: str
75 @param port: Default port, or 7531 if not given.
76 @type port: int
77 @param use_ssl: Whether to use SSL, or True if not given. Note that
78 there is no syntax in the connection string for specifying
79 whether or not to use SSL; if you want to control this you
80 will have to provide another method.
81 @type use_ssl: bool
82
83 @rtype: L{PBConnectionInfo}
84 """
85 auth = fpb.Authenticator(username=username, password=password)
86 ret = PBConnectionInfo(None, port, use_ssl, auth)
87
88 matched = _pat.search(string)
89 if not matched:
90 raise TypeError('Invalid connection string: %s '
91 '(looking for [user[:pass]@]host[/ssl|/tcp][:port])'
92 % string)
93
94 groups = matched.groups()
95 for o, k, i, f in ((auth, 'username', 1, str),
96 (auth, 'password', 3, str),
97 (ret, 'host', 4, str),
98 (ret, 'port', 6, int)):
99 if groups[i]:
100 setattr(o, k, f(groups[i]))
101 return ret
102