Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Demo Library kopdar python-ID Jogja 240916 untuk berinteraksi dengan zimbra API dengan dependensi module requests
__author__ = ('Imam Omar Mochtar', ('omar@jabetto.com', 'iomarmochtar@gmail.com'))
__version__ = (0 ,4)
"""
Dokumentasi zimbra soap berdasarkan versinyas
http://wiki.zimbra.com/wiki/SOAP_API_Reference_Material_Beginning_with_ZCS_8.0
"""
"""
Changelogs
==========
0.4:
- Management COS (getAll, create, delete, modify)
- Disable warnings https
- menghapus account (DeleteAccount)
0.3:
- pada ZMAIL otomatis set iszadmin menjadi false
0.2:
- memambahkan exception
- menambahkan ca => CreateAccount
0.1:
- Mengubah menjadi zimbra lib
- Menambahkan class ZMAIL untuk menerima dan mengirimkan email
"""
import requests
import json
import os
from pprint import pprint
from ConfigParser import ConfigParser
requests.packages.urllib3.disable_warnings()
class ZLoginFailed(Exception): pass
class ZConnectionErr(Exception): pass
class ZCommonErr(Exception): pass
class JZSOAP(object):
"""
Base SOAP
"""
token = None
soapurl = None
isdebug = False
username = None
password = None
def __init__(self, username=None, password=None, soapurl=None, isdebug=False, iszadmin=True, timeout=None, zinstance=None):
if not zinstance and None in (username, password, soapurl):
raise ZCommonErr("You must specified username, password and soapurl or zinstance to initiate object")
self.isdebug = isdebug
self.soapurl = soapurl
self.iszadmin = iszadmin
self.timeout = timeout
# jika ada instance dari zobjek maka set attribute saja
if zinstance:
self.setAttrs(zinstance)
else:
self.__login(username, password)
def __login(self, username, password):
"""
Login untuk mendapatkan token
"""
soapname = "Auth"
if self.iszadmin:
urn = "urn:zimbraAdmin"
body = {
"name": [{
"_content": username
}],
"password": [{
"_content": password
}]
}
else:
urn = "urn:zimbraAccount"
body = {
"account": {
"by": "name",
"_content": username
},
"password": {
"_content": password
}
}
result = self.send(soapname, body, urn)
if not result:
raise ZLoginFailed("Login Failed")
self.token = result['authToken'][0]['_content']
def getAttrs(self):
"""
Mendapatkan attribute yang akan didapatkan oleh objek
"""
return dict(
token = self.token,
soapurl = self.soapurl,
isdebug = self.isdebug,
username = self.username,
password = self.password
)
def setAttrs(self, zinstance):
"""
Set attribute dari zinstace yang diberikan pada current objek
"""
attrs = zinstace.getAttrs()
for k,v in attrs.iteritems():
setattr(self, k, v)
def printMe(self, word, isarray=False):
"""
Cetak verbose ke konsole
"""
if not self.isdebug: return
if not isarray:
print word
else:
pprint(word)
def send(self, soapname, body={}, urn="urn:zimbraAdmin"):
"""
Mengirimkan data dalam bentuk JSON, dan menerima dalam bentuk json
"""
# tambahkan urn di body
body['_jsns'] = urn
sendjson = {
"Header": {
"context": {
"authToken": [{'_content':self.token} if self.token else {}],
"nosession": {},
"userAgent": {
"name": "zmsoap"
},
"_jsns": "urn:zimbra"
}
},
"Body": {"%sRequest"%soapname:body},
"_jsns": "urn:zimbraSoap"
}
self.printMe('\nRequest BEGIN\n')
self.printMe(sendjson, isarray=True)
self.printMe('\nRequest END\n')
kwargs = {'verify': False}
if self.timeout:
kwargs['timeout'] = self.timeout
try:
result = requests.post(self.soapurl, json.dumps(sendjson), **kwargs).json()
except requests.exceptions.ConnectionError, e:
raise ZConnectionErr("Connection error: {0}".format(e))
body = result['Body']
# jika terjadi kesalahan
if body.has_key('Fault'):
self.printMe(body, True)
return False
# balikan sesuai dengan response yang diminta
self.printMe('\nResponse BEGIN\n')
self.printMe(body, isarray=True)
self.printMe('\nResponse END\n')
balik = body['%sResponse'%soapname]
return balik
class ZMAIL(JZSOAP):
"""
Mengirimkan dan menerima email (mailbox punya).
"""
def __init__(self, **kwargs):
if kwargs.has_key('iszadmin'):
del kwargs['iszadmin']
JZSOAP.__init__(self, iszadmin=False, **kwargs)
def sendMail(self, to, subject, body):
"""
Mengirimkan email
"""
soapname = "SendMsg"
urn = "urn:zimbraMail"
body = {
"m":{
"e":[
{
"t": "t",
"a": to,
"add": "0"
},
],
"su":{
"_content": subject
},
"mp":[
{
"ct":"text/plain",
"content":{
"_content": body
}
}
]
}
}
return self.send(soapname, body, urn)
def getMailbox(self, query="in:\"inbox\"", offset=0, limit=1):
"""
Mendapatkan isi mailbox
"""
soapname = "Search"
urn = "urn:zimbraMail"
body = {
"sortBy": "dateDesc",
"tz": {
"id": "Asia/Bangkok"
},
"locale": {
"_content": "en_US"
},
"offset": offset,
"limit": limit,
"query": query,
"types": "conversation",
"fetch": 1,
"html": 1
}
return self.send(soapname, body, urn)
class ZMPROV(JZSOAP):
"""
wrapper untuk zmprov command line yang menggunakan soap
"""
def gadl(self):
"""
Wrap seakan-akan menggunakan zmprov gadl
"""
exclude_dist = ['zimbradladmins@z.ong', 'zimbradomainadmins@z.ong']
head = 'GetAllDistributionLists'
data = {}
result = self.send(head, data)
return [x['name'] for x in result['dl'] if x['name'] not in exclude_dist ]
"""
# gdlm(distname, searchdata['search'], searchdata['page'])
def gdlm(self, distname, search, page):
head = 'GetDistributionList'
data = {
"dl": [{
"by": "name",
"_content": distname
}]
}
urn = "urn:zimbraAdmin"
members = []
result = self.send(head, data, urn)
try:
dlmlist = result['dl'][0]['dlm']
return paginate(dlmlist, '_content', search, page)
# return [ x['_content'] for x in result['dl'][0]['dlm'] ]
except:
return None
"""
def rdlm(self, distid, members):
"""
Menghapus distribution list member
"""
head = 'RemoveDistributionListMember'
data = {
'id': distid,
'dlm': [{'_content':x} for x in members ]
}
return self.send(head, data)
def adlm(self, distid, members):
"""
Menambahkan distribution list member
"""
head = 'AddDistributionListMember'
data = {
'id': distid,
'dlm': [{'_content':x} for x in members ]
}
return self.send(head, data)
def ra(self, accountid, newname):
"""
Rename account
"""
data = {
'id': accountid,
'newName': newname
}
return self.send('RenameAccount', data)
def ma(self, accountid, attribute, value=None):
"""
Modifikasi attribute user
"""
data = {
'id': accountid,
'a':{'n': attribute, '_content': value}
}
# jika tidak ada value berarti akan dihapus
if not value: del data['a']['_content']
return self.send('ModifyAccount', data)
def aaa(self, accountid, newalias):
"""
Menambahkan alias
"""
data = {
'id': accountid,
'alias': newalias
}
return self.send('AddAccountAlias', data)
def raa(self, accountid, alias):
"""
Menghapus account alias
"""
data = {
"id": accountid,
"alias": alias
}
return self.send('RemoveAccountAlias', data)
def gdlcheck(self, distname):
"""
Get distribution list id
"""
head = 'GetDistributionList'
data = {
"dl": [{
"by": "name",
"_content": distname
}]
}
result = self.send(head, data)
if result:
return result['dl'][0]['id']
return False
def ga(self, account):
"""
get account
"""
head = 'GetAccount'
data = {
'account': {
'by': 'name',
"_content": account
}
}
result = self.send(head, data)
if result:
return result['account'][0]['a']
return False
def cdl(self, distname):
"""
Membuat distribution list baru
"""
head = 'CreateDistributionList'
data = {
'name': distname,
'a': {
"n": "zimbraMailStatus",
}
}
result = self.send(head, data)
if result:
return True
return False
def ddl(self, distname):
"""
Menghapus distribution list
"""
# dapatkan dulu id dari distribution list yang dimaksud
dlid = self.gdlcheck(distname)
if not dlid:
return False
head = 'DeleteDistributionList'
data = {
'id': dlid
}
result = self.send(head, data)
if result:
return True
return False
def ca(self, name, password, attrs=[]):
"""
create account
"""
head = "CreateAccount"
data = {
"name": name,
"password": password,
"a": attrs
}
return self.send(head, data)
def da(self, name):
"""
Menhapus user
"""
acc = self.ga(name)
if not acc:
raise ZCommonErr("User {0} not found".format(name))
zimbraId = None
for data in acc:
if data['n'] != 'zimbraId': continue
zimbraId = data['_content']
break
head = "DeleteAccount"
data = {
"id": zimbraId
}
return self.send(head, data)
def gac(self, shortd=False):
"""
Mendapatkan semua cos yang ada
@shortd: Tampilkan hanya data sekilas saja
"""
result = self.send("GetAllCos")
if not shortd: return result
take = []
for x in result['cos']:
del x['a']
take.append( x )
return take
def mc(self, cosid, attr, value):
"""
Modify COS
"""
data = {
'id': [{
'_content': cosid
}],
'a':{'n': attr, '_content': value}
}
return self.send("ModifyCos", data)
def gc(self, cosid, attrs=None):
"""
Mendapatkan COS berdasarkan ID yang dimasukan
"""
params = {
"cos": {
"by": "id",
"_content": cosid
}
}
if attrs:
params["attrs"] = attrs
return self.send("GetCos", params)
# untuk getAllAccount berdasarkan name
def gaa(self, search='', uid=False, limit=0, offset=0):
"""
Mendapatkan semua user
"""
head = 'SearchDirectory'
data = {
'sortBy': 'name',
'limit': limit,
'offset' : offset,
'sortAscending': 1,
'attrs' : '',
# hanya mendapatkan normal account
"query": "(&(!(zimbraIsSystemAccount=TRUE)))"
}
if search:
data['query'] = 'mail=*{0}*'.format(search)
members = []
result = self.send(head, data)
if not result:
return []
if result.has_key('account'):
for member in result['account']:
if uid:
data = {
'name': member['name'],
'uid': member['id']
}
else:
data = member['name']
members.append(data)
return members
# untuk prevpage
# if page * settings.TOTAL_CONTENT_PER_PAGE > settings.TOTAL_CONTENT_PER_PAGE:
# prevpage = False
# else:
# prevpage = True
# nextpage = not result['more']
# if result['searchTotal'] == 0 or len(members) == 0:
# prevpage = True
# nextpage = True
# return {'members': members, 'nextpage': nextpage, 'prevpage': prevpage }
class ZMailBox(JZSOAP):
def getMbox(self, mboxid):
"""
Mendapatkan mailbox suatu akun
"""
return self.send(head, data)
if __name__ == '__main__':
zmprov = ZMPROV(
username='admin',
password='passwd123',
soapurl='https://127.0.0.1:7071/service/admin/soap',
# isdebug=True
)
pprint( zmprov.da("omar@mail.local") )
@iomarmochtar

This comment has been minimized.

Copy link
Owner Author

iomarmochtar commented Jun 12, 2017

i have create it's own repository for further development.

https://github.com/iomarmochtar/ozpy

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.