Using "Value::named" search for call to "os.path.join" not working

[Iceware]

CodeQL version 2.1.0

from rest_framework.viewsets import ModelViewSet
from rest_framework.exceptions import ValidationError
import os
from rest_framework.decorators import list_route


def func1():
    pass


def preprocess(request):
    pass


class TestViewSet(ModelViewSet):
    base = "111111"
    @list_route(methods=['post'])
    def upload(self, request):
        try:
            var1 = func1()
        except:
            raise ValidationError({'message': [_('1,1')]})
        if var1['1']:
            raise ValidationError({'message': [_('1,1')]})

        (upload_file, file_name) = preprocess(request)
        path1 = base.RELATED_DIR['11111111']
        if not os.path.exists(path1):
            os.makedirs(path1)
        path2 = os.path.join(path1, file_name)

Query:

import python
from ControlFlowNode c
where Value::named("os.path.join").getACall() = c
select c

Expect result:
Can find the call to "os.path.join" on the last line

Actual result:
Just show 2 call in the python lib

Is this a bug or am I doing this in the wrong way?

[tausbn]

Hi, my apologies for not replying to this sooner.

As far as I can see, everything you're doing is correct. I will try to see if I can reproduce this problem locally and figure out why the call isn't picked up correctly. 🤔

[tausbn]

So I spent some time debugging this, and I have an explanation for the behaviour you're seeing.

First, to pinpoint exactly where the problem lies, I tried inserting calls of the form os.path.join("foo") at various places in the source code, and then reran your query to see which of these were picked up. This revealed that everything below

        if var1['1']:

was missing from the analysis.

To figure out what's going on, let's first look at the possible values for var1:

import python

from NameNode n, Value v
where n.pointsTo(v) and n.getId() = "var1"
select n, v

This results in the following output

| test.py:24:12:24:15 | ControlFlowNode for var1 | file://:0:0:0:0 | None |

This means our analysis has inferred that var1 will have the value None on line 24 of the input file (which I named test.py).

Now, here's the issue: For our analysis, asking for a subscripted value of None is nonsensical -- None does not allow subscripting. This in turn means that our analysis cannot assign any value to the expression var1['1']. Because of this, the analysis assumes everything that follows is unreachable, and prunes it from further analysis. This is not entirely unreasonable, since if you were to attempt to run the above code, it would throw an exception at the var['1'] line (or possibly earlier), and so the code that follows is indeed not executed.

If instead we try changing the implementation of func1 to, say, return list_route, then the value of var1 is unknown, and so the analysis concludes that neither branch of the if can be pruned (and hence proceeds with analysing and locating the call to os.path.join that you were interested in).

Note that this depends somewhat on your local setup. If the import of rest_framework cannot be resolved, the analysis treats it as a "missing module", and treats any attributes of this module as having some "known unknown" value. If the import is resolved correctly, then the behaviour depends on the implementation of list_route. (After all, the module may set it to some value that isn't subscriptable, in which case we may end up in the same situation as above.)

I hope this explains the behaviour you encountered!

[Iceware]

Thanks very much for the detailed explain! This piece of code is from one of my old blog. The code is really old and complex. So I extracted more code and change the list_route to api_view as I am current using the latest version of django rest.

# coding=utf-8
import os

from os.path import join
from rest_framework.viewsets import ModelViewSet
from rest_framework.response import Response
from rest_framework.decorators import api_view
from rest_framework.exceptions import ValidationError

GLOBAL1 = "type"
GLOBAL2 = "filename"


class ClassA():
    def __init__(self):
        self.a_config = {"1": 1, "2": 2, "3": 3, "is_expired": True}

    def is_expired(self):
        return False


class ViewSetA(ModelViewSet):

    @api_view(['GET', 'POST'])
    def upload(self, request):
        try:
            info = func_1()
        except:
            raise ValidationError({'message': [_('message1')]})
        if info['is_expired']:
            raise ValidationError({'message': [_('message2')]})

        (var1, var2) = _preprocess(request)
        path1 = '/var/tmp'
        if not os.path.exists(path1):
            os.makedirs(path1)
        file_path = os.path.join(path1, var2)
        file_path = join(path1, var2)


def func_1():
    a = ClassA()
    info = a.a_config
    info.pop("1")
    info["is_expired"] = a.is_expired()
    return info


def _preprocess(request, type=GLOBAL1, pkg_type='linux'):
    try:
        upload_file = request.FILES['file']
    except:
        raise ValidationError({'message': [_('message2')]})

    file_name = upload_file.name.lower()
    if type == GLOBAL1:
        if not file_name.endswith('.zip'):
            raise ValidationError({'message': [_('message3')]})

        file_name = GLOBAL2
    if upload_file.size > 200 * 1024 * 1024:
        raise ValidationError({'message': [_('message4')]})
    return (upload_file, file_name)

Now the func_1 returns a dict and the api_view can be correctly resolved. The query in the beginning can correctly find "file_path = join(path1, var2)", but the query still cannot find the " file_path = os.path.join(path1, var2)"